Config

lume_services.config

Attributes

context module-attribute

context: containers.DeclarativeContainer = None

Classes

Context

Bases: containers.DeclarativeContainer

Attributes

config class-attribute

config = providers.Configuration()

mounted_filesystem class-attribute

mounted_filesystem = providers.Dependency(
    instance_of=MountedFilesystem, default=None
)

local_filesystem class-attribute

local_filesystem = providers.Singleton(LocalFilesystem)

model_db class-attribute

model_db = providers.Dependency(ModelDB)

results_db class-attribute

results_db = providers.Dependency(instance_of=ResultsDB)

scheduling_backend class-attribute

scheduling_backend = providers.Dependency(
    instance_of=Backend
)

file_service class-attribute

file_service = providers.Singleton(
    FileService,
    filesystems=providers.List(
        local_filesystem, mounted_filesystem
    ),
)

model_db_service class-attribute

model_db_service = providers.Singleton(
    ModelDBService, model_db=model_db
)

results_db_service class-attribute

results_db_service = providers.Singleton(
    ResultsDBService, results_db=results_db
)

scheduling_service class-attribute

scheduling_service = providers.Singleton(
    SchedulingService, backend=scheduling_backend
)

wiring_config class-attribute

wiring_config = containers.WiringConfiguration(
    packages=[
        "lume_services.files",
        "lume_services.results",
        "lume_services.flows",
    ]
)

LUMEServicesSettings

Bases: BaseSettings

Settings describing configuration for default LUME-services provider objects.

Attributes

model_db class-attribute

model_db: Optional[MySQLModelDBConfig]

results_db class-attribute

results_db: Optional[MongodbResultsDBConfig]

prefect class-attribute

prefect: Optional[PrefectConfig]

mounted_filesystem class-attribute

mounted_filesystem: Optional[MountedFilesystem]

backend class-attribute

backend: str = 'local'

Classes

Config

Attributes

validate_assignment class-attribute

validate_assignment = True

env_prefix class-attribute

env_prefix = 'LUME_'

env_nested_delimiter class-attribute

env_nested_delimiter = '__'

Functions

configure

configure(settings: LUMEServicesSettings = None)

Configure method with default methods for lume-services using MySQLModelDB and MongodbResultsDB.

Source code in lume_services/config.py

def configure(settings: LUMEServicesSettings = None):
    """Configure method with default methods for lume-services using MySQLModelDB
    and MongodbResultsDB.

    """
    if settings is None:
        try:
            settings = LUMEServicesSettings()

        except ValidationError as e:
            raise EnvironmentNotConfiguredError(
                list_env_vars(LUMEServicesSettings.schema()), validation_error=e
            )

    # apply prefect config
    if settings.prefect is not None:
        settings.prefect.apply()

    global context, _settings
    model_db = None
    if settings.model_db is not None:
        model_db = MySQLModelDB(settings.model_db)

    results_db = None
    if settings.results_db is not None:
        results_db = MongodbResultsDB(settings.results_db)

    # this could be moved to an enum
    if settings.backend is not None:
        if settings.backend == "kubernetes":
            backend = KubernetesBackend(config=settings.prefect)

        elif settings.backend == "docker":
            backend = DockerBackend(config=settings.prefect)

        elif settings.backend == "local":
            backend = LocalBackend()

        else:
            raise ValueError(f"Unsupported backend {settings.backend}")

    # default to local
    else:
        backend = LocalBackend()

    context = Context(
        model_db=model_db,
        results_db=results_db,
        mounted_filesystem=settings.mounted_filesystem,
        scheduling_backend=backend,
    )
    _settings = settings

list_env_vars

list_env_vars(
    schema: dict = LUMEServicesSettings.schema(),
    prefix: str = LUMEServicesSettings.Config.env_prefix,
    delimiter: str = LUMEServicesSettings.Config.env_nested_delimiter,
) -> list

Source code in lume_services/config.py

def list_env_vars(
    schema: dict = LUMEServicesSettings.schema(),
    prefix: str = LUMEServicesSettings.Config.env_prefix,
    delimiter: str = LUMEServicesSettings.Config.env_nested_delimiter,
) -> list:
    env_vars = {"base": []}

    def unpack_props(
        props,
        parent,
        env_vars=env_vars,
        prefix=prefix,
        delimiter=delimiter,
        schema=schema,
    ):

        for prop_name, prop in props.items():
            if "properties" in prop:
                unpack_props(
                    prop["properties"], prefix=f"{prefix}{delimiter}{prop_name}"
                )

            elif "allOf" in prop:

                sub_prop_reference = prop["allOf"][0]["$ref"]
                # prepare from format #/
                sub_prop_reference = sub_prop_reference.replace("#/", "")
                sub_prop_reference = sub_prop_reference.split("/")
                reference_locale = schema
                for reference in sub_prop_reference:
                    reference_locale = reference_locale[reference]

                unpack_props(
                    reference_locale["properties"],
                    parent=parent,
                    prefix=f"{prefix}{delimiter}{prop_name}",
                )

            else:
                if "env_names" in prop:
                    prop_names = list(prop["env_names"])
                    env_vars[parent] += [
                        f"{prefix}{delimiter}{name}".upper() for name in prop_names
                    ]
                else:
                    env_vars[parent].append(f"{prefix}{delimiter}{prop_name}".upper())

    # iterate over top level definitions
    for item_name, item in schema["properties"].items():

        env_name = list(item["env_names"])[0]

        if "allOf" in item:

            sub_prop_reference = item["allOf"][0]["$ref"]
            # prepare from format #/
            sub_prop_reference = sub_prop_reference.replace("#/", "")
            sub_prop_reference = sub_prop_reference.split("/")
            reference_locale = schema
            for reference in sub_prop_reference:
                reference_locale = reference_locale[reference]

            env_vars[item_name] = []
            unpack_props(
                reference_locale["properties"], prefix=env_name, parent=item_name
            )

        else:
            env_vars["base"].append(env_name.upper())

    # capitalize

    return env_vars