Brian Bjarke Jensen
2ee5ab2b9c
CI Pipeline / Test (pull_request) Successful in 2m7s
CI Pipeline / Build and Publish (./Dockerfile.model, ${{ vars.docker_repo_url }}/${{ gitea.repository }}/model) (pull_request) Successful in 16s
CI Pipeline / Build and Publish (./Dockerfile.web_ui, ${{ vars.docker_repo_url }}/${{ gitea.repository }}/web_ui) (pull_request) Successful in 15s
46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
"""Definition of connect function."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
from minio import Minio
|
|
|
|
|
|
def connect(
|
|
) -> Minio:
|
|
"""Connect to MinIO server"""
|
|
# ensure necessary env vars available
|
|
necesasary_var_list = [
|
|
'MINIO_ENDPOINT',
|
|
'MINIO_ACCESS_KEY',
|
|
'MINIO_SECRET_KEY',
|
|
'MINIO_BUCKET_NAME',
|
|
]
|
|
for env_var in necesasary_var_list:
|
|
# ensure env var set
|
|
assert (
|
|
env_var in os.environ
|
|
), (
|
|
f"environment variable not set: {env_var}"
|
|
)
|
|
# prepare arguments
|
|
minio_endpoint = os.getenv('MINIO_ENDPOINT', default=None)
|
|
assert isinstance(minio_endpoint, str)
|
|
minio_access_key = os.getenv('MINIO_ACCESS_KEY', default=None)
|
|
assert isinstance(minio_access_key, str)
|
|
minio_secret_key = os.getenv('MINIO_SECRET_KEY', default=None)
|
|
assert isinstance(minio_secret_key, str)
|
|
minio_bucket_name = os.getenv('MINIO_BUCKET_NAME', default=None)
|
|
assert isinstance(minio_bucket_name, str)
|
|
# 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):
|
|
client.make_bucket(bucket_name=minio_bucket_name)
|
|
return client
|