34 lines
950 B
Python
34 lines
950 B
Python
import re
|
|
|
|
from minio import Minio
|
|
|
|
from pyinfra.config import CONFIG
|
|
from pyinfra.exceptions import InvalidEndpoint
|
|
|
|
|
|
def parse_endpoint(endpoint):
|
|
endpoint_pattern = r"(?P<protocol>https?)://(?P<address>(?:(?:(?:\d{1,3}\.){3}\d{1,3})|(?:\w|\.)+)(?:\:\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)
|