36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
import logging
|
|
import re
|
|
|
|
from minio import Minio
|
|
|
|
from pyinfra.config import CONFIG
|
|
from pyinfra.exceptions import InvalidEndpoint
|
|
|
|
|
|
def parse_endpoint(endpoint):
|
|
# FIXME Greedy matching (.+) since we get random storage names on kubernetes (eg http://red-research-headless:9000)
|
|
endpoint_pattern = r"(?P<protocol>https?)*(?:://)*(?P<address>(?:(?:(?:\d{1,3}\.){3}\d{1,3})|.+)(?:\:\d+)?)"
|
|
|
|
match = re.match(endpoint_pattern, endpoint)
|
|
|
|
if not match:
|
|
raise InvalidEndpoint(f"Endpoint {endpoint} is invalid; expected {endpoint_pattern}")
|
|
|
|
return {"secure": match.group("protocol") == "https", "endpoint": match.group("address")}
|
|
|
|
|
|
def get_s3_client(params=None) -> Minio:
|
|
"""
|
|
Args:
|
|
params: dict like
|
|
{
|
|
"endpoint": <storage_endpoint>
|
|
"access_key": <storage_key>
|
|
"secret_key": <storage_secret>
|
|
}
|
|
"""
|
|
if not params:
|
|
params = CONFIG.storage.s3
|
|
|
|
return Minio(**parse_endpoint(params.endpoint), access_key=params.access_key, secret_key=params.secret_key)
|