Files
spoolman2/spoolman/env.py
2023-08-28 21:10:07 +02:00

248 lines
7.1 KiB
Python

"""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
import pkg_resources
from platformdirs import user_data_dir
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 in {"FALSE", "0"}:
return False
if debug_mode in {"TRUE", "1"}:
return True
raise ValueError(f"Failed to parse SPOOLMAN_DEBUG_MODE variable: Unknown debug mode '{debug_mode}'.")
def is_automatic_backup_enabled() -> bool:
"""Get whether automatic backup is enabled from environment variables.
Returns True if no environment variable was set for automatic backup.
Returns:
bool: Whether automatic backup is enabled.
"""
automatic_backup = os.getenv("SPOOLMAN_AUTOMATIC_BACKUP", "TRUE").upper()
if automatic_backup in {"FALSE", "0"}:
return False
if automatic_backup in {"TRUE", "1"}:
return True
raise ValueError(
f"Failed to parse SPOOLMAN_AUTOMATIC_BACKUP variable: Unknown automatic backup '{automatic_backup}'.",
)
def get_data_dir() -> Path:
"""Get the data directory.
Returns:
Path: The data directory.
"""
data_dir = Path(user_data_dir("spoolman"))
data_dir.mkdir(parents=True, exist_ok=True)
return data_dir
def get_backups_dir() -> Path:
"""Get the backups directory.
Returns:
Path: The backups directory.
"""
data_dir = get_data_dir()
backups_dir = data_dir.joinpath("backups")
backups_dir.mkdir(parents=True, exist_ok=True)
return backups_dir
def get_version() -> str:
"""Get the version of the package.
Returns:
str: The version.
"""
return pkg_resources.get_distribution("spoolman").version