"""Utilities for grabbing config from environment variables.""" import logging import os from enum import Enum from pathlib import Path from typing import Optional from urllib import parse class DatabaseType(Enum): """The database type.""" POSTGRES = "postgres" MYSQL = "mysql" SQLITE = "sqlite" COCKROACHDB = "cockroachdb" def to_drivername(self: "DatabaseType") -> str: """Get the drivername for the database type. Returns: str: The drivername. """ if self is DatabaseType.POSTGRES: return "postgresql+asyncpg" if self is DatabaseType.MYSQL: return "mysql+aiomysql" if self is DatabaseType.SQLITE: return "sqlite+aiosqlite" if self is DatabaseType.COCKROACHDB: return "cockroachdb+asyncpg" raise ValueError(f"Unknown database type '{self}'.") def get_database_type() -> Optional[DatabaseType]: """Get the database type from environment variables. Returns None if no environment variable was set for the database type. Returns: Optional[DatabaseType]: The database type. """ database_type = os.getenv("SPOOLMAN_DB_TYPE") if database_type is None: return None if database_type == "postgres": return DatabaseType.POSTGRES if database_type == "mysql": return DatabaseType.MYSQL if database_type == "sqlite": return DatabaseType.SQLITE if database_type == "cockroachdb": return DatabaseType.COCKROACHDB raise ValueError(f"Failed to parse SPOOLMAN_DB_TYPE variable: Unknown database type '{database_type}'.") def get_host() -> Optional[str]: """Get the DB host from environment variables. Returns None if no environment variable was set for the host. Returns: Optional[str]: The host. """ return os.getenv("SPOOLMAN_DB_HOST") def get_port() -> Optional[int]: """Get the DB port from environment variables. Returns None if no environment variable was set for the port. Returns: Optional[str]: The port. """ port = os.getenv("SPOOLMAN_DB_PORT") if port is None: return None try: return int(port) except ValueError as exc: raise ValueError(f"Failed to parse SPOOLMAN_DB_PORT variable: {exc!s}") from exc def get_database() -> Optional[str]: """Get the DB name from environment variables. Returns None if no environment variable was set for the name. Returns: Optional[str]: The name. """ return os.getenv("SPOOLMAN_DB_NAME") def get_query() -> Optional[dict[str, str]]: """Get the DB query from environment variables. Returns None if no environment variable was set for the query. Returns: Optional[dict]: The query. """ query = os.getenv("SPOOLMAN_DB_QUERY") if query is None: return None try: parsed_dict = parse.parse_qs(query, strict_parsing=True) return {key: value[0] for key, value in parsed_dict.items()} except ValueError as exc: raise ValueError(f"Failed to parse SPOOLMAN_DB_QUERY variable: {exc!s}") from exc def get_username() -> Optional[str]: """Get the DB username from environment variables. Returns None if no environment variable was set for the username. Returns: Optional[str]: The username. """ return os.getenv("SPOOLMAN_DB_USERNAME") def get_password() -> Optional[str]: """Get the DB password from environment variables. Returns None if no environment variables were set for the password. Raises: ValueError: If it failed to read the password from a password file. Returns: Optional[str]: The password. """ # First attempt: grab password from a file. This is the most secure way of storing passwords. file_path = os.getenv("SPOOLMAN_DB_PASSWORD_FILE") if file_path is not None: file = Path(file_path) if not file.exists() or not file.is_file(): raise ValueError( "Failed to parse SPOOLMAN_DB_PASSWORD_FILE variable: " f'Database password file "{file_path}" does not exist.', ) try: with file.open(encoding="utf-8") as f: return f.read() except OSError as exc: raise ValueError( "Failed to parse SPOOLMAN_DB_PASSWORD_FILE variable: " f'Failed to read password from file "{file_path}": {exc!s}.', ) from exc # Second attempt: grab directly from an environment variable. return os.getenv("SPOOLMAN_DB_PASSWORD") def get_logging_level() -> int: """Get the logging level from environment variables. Returns "INFO" if no environment variable was set for the logging level. Returns: str: The logging level. """ log_level_str = os.getenv("SPOOLMAN_LOGGING_LEVEL", "INFO").upper() if log_level_str == "DEBUG": return logging.DEBUG if log_level_str == "INFO": return logging.INFO if log_level_str == "WARNING": return logging.WARNING if log_level_str == "ERROR": return logging.ERROR if log_level_str == "CRITICAL": return logging.CRITICAL raise ValueError(f"Failed to parse SPOOLMAN_LOGGING_LEVEL variable: Unknown logging level '{log_level_str}'.") def is_debug_mode() -> bool: """Get whether debug mode is enabled from environment variables. Returns False if no environment variable was set for debug mode. Returns: bool: Whether debug mode is enabled. """ debug_mode = os.getenv("SPOOLMAN_DEBUG_MODE", "FALSE").upper() if debug_mode == "FALSE" or debug_mode == "0": return False if debug_mode == "TRUE" or debug_mode == "1": return True raise ValueError(f"Failed to parse SPOOLMAN_DEBUG_MODE variable: Unknown debug mode '{debug_mode}'.")