opsml.settings.config

View Source

 1# Copyright (c) Shipt, Inc.
 2# This source code is licensed under the MIT license found in the
 3# LICENSE file in the root directory of this source tree.
 4
 5import re
 6from pathlib import Path
 7from typing import Optional
 8
 9from pydantic import field_validator
10from pydantic_settings import BaseSettings
11
12from opsml.types import StorageSystem
13
14
15class OpsmlConfig(BaseSettings):
16    app_name: str = "opsml"
17    app_env: str = "development"
18
19    opsml_storage_uri: str = "./mlruns"
20    opsml_tracking_uri: str = "sqlite:///tmp.db"
21    opsml_prod_token: str = "staging"
22    opsml_proxy_root: str = "opsml-root:/"
23    opsml_registry_path: str = "model_registry"
24    opsml_testing: bool = bool(0)
25    download_chunk_size: int = 31457280  # 30MB
26    upload_chunk_size: int = 31457280  # 30MB
27
28    # API client username / password
29    opsml_username: Optional[str] = None
30    opsml_password: Optional[str] = None
31
32    # The current RUN_ID to load when creating a new project
33    opsml_run_id: Optional[str] = None
34
35    @field_validator("opsml_storage_uri", mode="before")
36    @classmethod
37    def set_opsml_storage_uri(cls, opsml_storage_uri: str) -> str:
38        """Opsml uses storage cients that follow fsspec guidelines. LocalFileSytem only deals
39        in absolutes, so we need to convert relative paths to absolute paths.
40        """
41        if opsml_storage_uri.startswith("gs://") or opsml_storage_uri.startswith("s3://"):
42            return opsml_storage_uri
43
44        return Path(opsml_storage_uri).absolute().as_posix()
45
46    @property
47    def is_tracking_local(self) -> bool:
48        """Used to determine if an API client will be used.
49
50        If tracking is local, the [server] extra is required.
51        """
52        return not self.opsml_tracking_uri.lower().strip().startswith("http")
53
54    @property
55    def storage_system(self) -> StorageSystem:
56        """Returns the storage system used for the current tracking URI"""
57        if self.is_tracking_local:
58            if self.opsml_storage_uri.startswith("gs://"):
59                return StorageSystem.GCS
60            if self.opsml_storage_uri.startswith("s3://"):
61                return StorageSystem.S3
62            return StorageSystem.LOCAL
63        return StorageSystem.API
64
65    @property
66    def storage_root(self) -> str:
67        """Returns the root of the storage URI"""
68        if self.is_tracking_local:
69            storage_uri_lower = self.opsml_storage_uri.lower()
70            if storage_uri_lower.startswith("gs://"):
71                return re.sub("^gs://", "", storage_uri_lower)
72            if storage_uri_lower.startswith("s3://"):
73                return re.sub("^s3://", "", storage_uri_lower)
74            return storage_uri_lower
75        return self.opsml_proxy_root
76
77
78config = OpsmlConfig()

class OpsmlConfig(pydantic_settings.main.BaseSettings): View Source

16class OpsmlConfig(BaseSettings):
17    app_name: str = "opsml"
18    app_env: str = "development"
19
20    opsml_storage_uri: str = "./mlruns"
21    opsml_tracking_uri: str = "sqlite:///tmp.db"
22    opsml_prod_token: str = "staging"
23    opsml_proxy_root: str = "opsml-root:/"
24    opsml_registry_path: str = "model_registry"
25    opsml_testing: bool = bool(0)
26    download_chunk_size: int = 31457280  # 30MB
27    upload_chunk_size: int = 31457280  # 30MB
28
29    # API client username / password
30    opsml_username: Optional[str] = None
31    opsml_password: Optional[str] = None
32
33    # The current RUN_ID to load when creating a new project
34    opsml_run_id: Optional[str] = None
35
36    @field_validator("opsml_storage_uri", mode="before")
37    @classmethod
38    def set_opsml_storage_uri(cls, opsml_storage_uri: str) -> str:
39        """Opsml uses storage cients that follow fsspec guidelines. LocalFileSytem only deals
40        in absolutes, so we need to convert relative paths to absolute paths.
41        """
42        if opsml_storage_uri.startswith("gs://") or opsml_storage_uri.startswith("s3://"):
43            return opsml_storage_uri
44
45        return Path(opsml_storage_uri).absolute().as_posix()
46
47    @property
48    def is_tracking_local(self) -> bool:
49        """Used to determine if an API client will be used.
50
51        If tracking is local, the [server] extra is required.
52        """
53        return not self.opsml_tracking_uri.lower().strip().startswith("http")
54
55    @property
56    def storage_system(self) -> StorageSystem:
57        """Returns the storage system used for the current tracking URI"""
58        if self.is_tracking_local:
59            if self.opsml_storage_uri.startswith("gs://"):
60                return StorageSystem.GCS
61            if self.opsml_storage_uri.startswith("s3://"):
62                return StorageSystem.S3
63            return StorageSystem.LOCAL
64        return StorageSystem.API
65
66    @property
67    def storage_root(self) -> str:
68        """Returns the root of the storage URI"""
69        if self.is_tracking_local:
70            storage_uri_lower = self.opsml_storage_uri.lower()
71            if storage_uri_lower.startswith("gs://"):
72                return re.sub("^gs://", "", storage_uri_lower)
73            if storage_uri_lower.startswith("s3://"):
74                return re.sub("^s3://", "", storage_uri_lower)
75            return storage_uri_lower
76        return self.opsml_proxy_root

Base class for settings, allowing values to be overridden by environment variables.

This is useful in production for secrets you do not wish to save in code, it plays nicely with docker(-compose), Heroku and any 12 factor app design.

All the below attributes can be set via model_config.

Arguments:

_case_sensitive: Whether environment variables names should be read with case-sensitivity. Defaults to None.
_env_prefix: Prefix for all environment variables. Defaults to None.
_env_file: The env file(s) to load settings values from. Defaults to Path(''), which means that the value from model_config['env_file'] should be used. You can also pass None to indicate that environment variables should not be loaded from an env file.
_env_file_encoding: The env file encoding, e.g. 'latin-1'. Defaults to None.
_env_ignore_empty: Ignore environment variables where the value is an empty string. Default to False.
_env_nested_delimiter: The nested env values delimiter. Defaults to None.
_env_parse_none_str: The env string value that should be parsed (e.g. "null", "void", "None", etc.) into None type(None). Defaults to None type(None), which means no parsing should occur.
_secrets_dir: The secret files directory. Defaults to None.

app_name: str

app_env: str

opsml_storage_uri: str

opsml_tracking_uri: str

opsml_prod_token: str

opsml_proxy_root: str

opsml_registry_path: str

opsml_testing: bool

download_chunk_size: int

upload_chunk_size: int

opsml_username: Optional[str]

opsml_password: Optional[str]

opsml_run_id: Optional[str]

@field_validator('opsml_storage_uri', mode='before')

@classmethod

def set_opsml_storage_uri(cls, opsml_storage_uri: str) -> str: View Source

36    @field_validator("opsml_storage_uri", mode="before")
37    @classmethod
38    def set_opsml_storage_uri(cls, opsml_storage_uri: str) -> str:
39        """Opsml uses storage cients that follow fsspec guidelines. LocalFileSytem only deals
40        in absolutes, so we need to convert relative paths to absolute paths.
41        """
42        if opsml_storage_uri.startswith("gs://") or opsml_storage_uri.startswith("s3://"):
43            return opsml_storage_uri
44
45        return Path(opsml_storage_uri).absolute().as_posix()

Opsml uses storage cients that follow fsspec guidelines. LocalFileSytem only deals in absolutes, so we need to convert relative paths to absolute paths.

is_tracking_local: bool View Source

47    @property
48    def is_tracking_local(self) -> bool:
49        """Used to determine if an API client will be used.
50
51        If tracking is local, the [server] extra is required.
52        """
53        return not self.opsml_tracking_uri.lower().strip().startswith("http")

Used to determine if an API client will be used.

If tracking is local, the [server] extra is required.

storage_system: opsml.types.storage.StorageSystem View Source

55    @property
56    def storage_system(self) -> StorageSystem:
57        """Returns the storage system used for the current tracking URI"""
58        if self.is_tracking_local:
59            if self.opsml_storage_uri.startswith("gs://"):
60                return StorageSystem.GCS
61            if self.opsml_storage_uri.startswith("s3://"):
62                return StorageSystem.S3
63            return StorageSystem.LOCAL
64        return StorageSystem.API

Returns the storage system used for the current tracking URI

storage_root: str View Source

66    @property
67    def storage_root(self) -> str:
68        """Returns the root of the storage URI"""
69        if self.is_tracking_local:
70            storage_uri_lower = self.opsml_storage_uri.lower()
71            if storage_uri_lower.startswith("gs://"):
72                return re.sub("^gs://", "", storage_uri_lower)
73            if storage_uri_lower.startswith("s3://"):
74                return re.sub("^s3://", "", storage_uri_lower)
75            return storage_uri_lower
76        return self.opsml_proxy_root

Returns the root of the storage URI

model_config: ClassVar[pydantic_settings.main.SettingsConfigDict] = {'extra': 'forbid', 'arbitrary_types_allowed': True, 'validate_default': True, 'case_sensitive': False, 'env_prefix': '', 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_parse_none_str': None, 'json_file': None, 'json_file_encoding': None, 'yaml_file': None, 'yaml_file_encoding': None, 'toml_file': None, 'secrets_dir': None, 'protected_namespaces': ('model_', 'settings_')}

model_fields = {'app_name': FieldInfo(annotation=str, required=False, default='opsml'), 'app_env': FieldInfo(annotation=str, required=False, default='development'), 'opsml_storage_uri': FieldInfo(annotation=str, required=False, default='./mlruns'), 'opsml_tracking_uri': FieldInfo(annotation=str, required=False, default='sqlite:///tmp.db'), 'opsml_prod_token': FieldInfo(annotation=str, required=False, default='staging'), 'opsml_proxy_root': FieldInfo(annotation=str, required=False, default='opsml-root:/'), 'opsml_registry_path': FieldInfo(annotation=str, required=False, default='model_registry'), 'opsml_testing': FieldInfo(annotation=bool, required=False, default=False), 'download_chunk_size': FieldInfo(annotation=int, required=False, default=31457280), 'upload_chunk_size': FieldInfo(annotation=int, required=False, default=31457280), 'opsml_username': FieldInfo(annotation=Union[str, NoneType], required=False), 'opsml_password': FieldInfo(annotation=Union[str, NoneType], required=False), 'opsml_run_id': FieldInfo(annotation=Union[str, NoneType], required=False)}

model_computed_fields = {}

Inherited Members

pydantic_settings.main.BaseSettings: BaseSettings; settings_customise_sources
pydantic.main.BaseModel: model_extra; model_fields_set; model_construct; model_copy; model_dump; model_dump_json; model_json_schema; model_parametrized_name; model_post_init; model_rebuild; model_validate; model_validate_json; model_validate_strings; dict; json; parse_obj; parse_raw; parse_file; from_orm; construct; copy; schema; schema_json; validate; update_forward_refs

config = OpsmlConfig(app_name='opsml', app_env='development', opsml_storage_uri='gs://opsml-integration', opsml_tracking_uri='sqlite:///tmp.db', opsml_prod_token='staging', opsml_proxy_root='opsml-root:/', opsml_registry_path='model_registry', opsml_testing=False, download_chunk_size=31457280, upload_chunk_size=31457280, opsml_username=None, opsml_password=None, opsml_run_id=None)