Split functions into separate files

great_ai/src/great_ai/great_ai/constants.py

@@ -0,0 +1,3 @@
+ENV_VAR_KEY = "ENVIRONMENT"
+PRODUCTION_KEY = "production"
+DEFAULT_TRACING_DB_FILENAME = "tracing_database.json"

great_ai/src/great_ai/great_ai/context/__init__.py

@@ -1,2 +1,3 @@
+from .configure import configure
 from .context import Context
-from .get_context import configure, get_context
+from .get_context import get_context

great_ai/src/great_ai/great_ai/context/configure.py

@@ -0,0 +1,91 @@
+import os
+import random
+from logging import INFO, Logger
+from pathlib import Path
+
+import great_ai.great_ai.context.context as context
+from great_ai.open_s3 import LargeFile
+from great_ai.utilities.logger import create_logger
+
+from ..constants import DEFAULT_TRACING_DB_FILENAME, ENV_VAR_KEY, PRODUCTION_KEY
+from ..persistence import ParallelTinyDbDriver, PersistenceDriver
+
+
+def configure(
+    log_level: int = INFO,
+    s3_config: Path = Path("s3.ini"),
+    seed: int = 42,
+    persistence_driver: PersistenceDriver = ParallelTinyDbDriver(
+        Path(DEFAULT_TRACING_DB_FILENAME)
+    ),
+) -> None:
+    logger = create_logger("great_ai", level=log_level)
+
+    if context._context is not None:
+        logger.warn(
+            "Configuration has been already initialised, overwriting.\n"
+            + 'Make sure to call "configure()" before importing your application code.'
+        )
+
+    is_production = _is_in_production_mode(
+        logger=logger,
+    )
+    _initialize_large_file(s3_config, logger=logger)
+    _set_seed(seed)
+
+    if not persistence_driver.is_threadsafe:
+        logger.warning(
+            f"The selected persistence driver ({type(persistence_driver).__name__}) is not threadsafe"
+        )
+
+    context._context = context.Context(
+        metrics_path="/metrics",
+        persistence=persistence_driver,
+        is_production=is_production,
+        logger=logger,
+    )
+
+    logger.info("Options: configured ✅")
+
+
+def _is_in_production_mode(logger: Logger) -> bool:
+    environment = os.environ.get(ENV_VAR_KEY)
+
+    if environment is None:
+        logger.info(
+            f"Environment variable {ENV_VAR_KEY} is not set, defaulting to development mode"
+        )
+        is_production = False
+    else:
+        is_production = environment.lower() == PRODUCTION_KEY
+        if not is_production:
+            logger.info(
+                f"Value of {ENV_VAR_KEY} is `{environment}` which is not equal to `{PRODUCTION_KEY}`"
+            )
+
+    if is_production:
+        logger.info("Running in production mode ✅")
+    else:
+        logger.warning("Running in development mode ‼️")
+
+    return is_production
+
+
+def _initialize_large_file(s3_config: Path, logger: Logger) -> None:
+    if s3_config.exists():
+        LargeFile.configure_credentials_from_file(s3_config)
+    else:
+        logger.warning(
+            f"Provided S3 config ({s3_config.resolve()}) not found, skipping LargeFile initialisation"
+        )
+
+
+def _set_seed(seed: int) -> None:
+    random.seed(seed)
+
+    try:
+        import numpy
+
+        numpy.random.seed(seed + 1)
+    except ImportError:
+        pass

great_ai/src/great_ai/great_ai/context/context.py

@@ -1,4 +1,5 @@
 from logging import Logger
+from typing import Optional
 
 from pydantic import BaseModel
 
@@ -13,3 +14,6 @@ class Context(BaseModel):
 
     class Config:
         arbitrary_types_allowed = True
+
+
+_context: Optional[Context] = None

great_ai/src/great_ai/great_ai/context/get_context.py

@@ -1,94 +1,12 @@
-import os
+from typing import cast
-import random
-from logging import INFO, Logger
-from pathlib import Path
-from typing import Optional, cast
 
-from great_ai.open_s3 import LargeFile
+import great_ai.great_ai.context.context as context
-from great_ai.utilities.logger import create_logger
 
-from ..persistence import ParallelTinyDbDriver, PersistenceDriver
+from .configure import configure
-from .context import Context
-
-_context: Optional[Context] = None
-PRODUCTION_KEY = "production"
 
 
-def get_context() -> Context:
+def get_context() -> context.Context:
-    if _context is None:
+    if context._context is None:
         configure()
 
-    return cast(Context, _context)
+    return cast(context.Context, context._context)
-
-
-def configure(
-    log_level: int = INFO,
-    s3_config: Path = Path("s3.ini"),
-    seed: int = 42,
-    persistence_driver: PersistenceDriver = ParallelTinyDbDriver(
-        Path("tracing_database.json")
-    ),
-    development_mode_override: Optional[bool] = None,
-) -> None:
-    global _context
-
-    logger = create_logger("great_ai", level=log_level)
-
-    if _context is not None:
-        logger.warn(
-            "Configuration has been already initialised, overwriting.\n"
-            + 'Make sure to call "configure()" before importing your application code.'
-        )
-
-    is_production = _is_in_production_mode(
-        override=None
-        if development_mode_override is None
-        else not development_mode_override,
-        logger=logger,
-    )
-    _initialize_large_file(s3_config, logger=logger)
-    _set_seed(seed)
-
-    if not persistence_driver.is_threadsafe:
-        logger.warning(
-            f"The selected persistence driver ({type(persistence_driver).__name__}) is not threadsafe"
-        )
-    _context = Context(
-        metrics_path="/metrics",
-        persistence=persistence_driver,
-        is_production=is_production,
-        logger=logger,
-    )
-
-    logger.info("Options: configured ✅")
-
-
-def _is_in_production_mode(override: Optional[bool], logger: Logger) -> bool:
-    environment = os.environ.get("ENVIRONMENT", PRODUCTION_KEY).lower()
-    is_production = environment == PRODUCTION_KEY if override is None else override
-
-    if is_production:
-        logger.info("Running in production mode ✅")
-    else:
-        logger.warning("Running in development mode ‼️")
-
-    return is_production
-
-
-def _initialize_large_file(s3_config: Path, logger: Logger) -> None:
-    if s3_config.exists():
-        LargeFile.configure_credentials_from_file(s3_config)
-    else:
-        logger.info(
-            f"Provided S3 config ({s3_config.resolve()}) not found, skipping LargeFile initialisation"
-        )
-
-
-def _set_seed(seed: int) -> None:
-    random.seed(seed)
-    try:
-        import numpy
-
-        numpy.random.seed(seed + 1)
-    except ImportError:
-        pass