39 lines
1.1 KiB
Python
39 lines
1.1 KiB
Python
"""Definition of connect function."""
|
|
|
|
import logging
|
|
import os
|
|
|
|
from minio import Minio
|
|
|
|
from shared.utils import check_env
|
|
|
|
|
|
def connect_minio() -> Minio:
|
|
"""Connect to MinIO server."""
|
|
# ensure necessary env vars available
|
|
var_list = {
|
|
'MINIO_ENDPOINT',
|
|
'MINIO_ACCESS_KEY',
|
|
'MINIO_SECRET_KEY',
|
|
'MINIO_BUCKET_NAME',
|
|
}
|
|
check_env(var_list)
|
|
# prepare arguments
|
|
minio_endpoint = str(os.getenv('MINIO_ENDPOINT'))
|
|
minio_access_key = str(os.getenv('MINIO_ACCESS_KEY'))
|
|
minio_secret_key = str(os.getenv('MINIO_SECRET_KEY'))
|
|
minio_bucket_name = str(os.getenv('MINIO_BUCKET_NAME'))
|
|
# connect client
|
|
client = Minio(
|
|
endpoint=minio_endpoint,
|
|
access_key=minio_access_key,
|
|
secret_key=minio_secret_key,
|
|
secure=False,
|
|
)
|
|
# ensure bucket exists
|
|
if not client.bucket_exists(bucket_name=minio_bucket_name):
|
|
logging.info('creating bucket: %s', minio_bucket_name)
|
|
client.make_bucket(bucket_name=minio_bucket_name)
|
|
logging.debug('finished')
|
|
return client
|