Add config structure and database init function

This commit is contained in:
Ethan Paul 2022-02-08 22:49:01 -05:00
parent bf78aeac98
commit b86a8e5083
No known key found for this signature in database
GPG Key ID: 6A337337DF6B5B1A
3 changed files with 174 additions and 2 deletions

111
section7/configuration.py Normal file
View File

@ -0,0 +1,111 @@
import os
from dataclasses import dataclass
from dataclasses import field
from pathlib import Path
from typing import Optional
from section7 import constants
@dataclass
class DatabaseSqliteConfig:
"""SQLite database backend configuration options
:param path: Path to the SQLite database file
"""
path: Path = Path.cwd() / "section7.db"
@classmethod
def from_env(cls):
"""Build dataclass from environment"""
return cls(
path=Path(os.environ.get("SECTION7_DATABASE_SQLITE_PATH", cls.path))
.expanduser()
.resolve(),
)
@dataclass
class DatabaseMariaConfig:
"""MariaDB database backend configuration options
:param hostname: Hostname or IP address of the host running the database server
:param username: Username of the account to use for connecting to the database server
:param password: Password for the account to use for connecting to the database server
:param port: Port on the host that the database server is listening on
:param schema: Database schema that the application should use
"""
hostname: str = "localhost"
username: str = "root"
password: Optional[str] = None
port: int = 3306
schema: str = "section7"
@classmethod
def from_env(cls):
"""Build dataclass from environment"""
return cls(
hostname=os.getenv("SECTION7_DATABASE_MARIADB_HOSTNAME", cls.hostname),
username=os.getenv("SECTION7_DATABASE_MARIADB_USERNAME", cls.username),
password=os.environ.get("SECTION7_DATABASE_MARIADB_PASSWORD", cls.password),
port=int(os.environ.get("SECTION7_DATABASE_MARIADB_PORT", cls.port)),
schema=os.getenv("SECTION7_DATABASE_MARIADB_SCHEMA", cls.schema),
)
@dataclass
class DatabaseConfig:
"""Database backend configuration
:param backend: Enum selecting the backend to use for storing data
:param sqlite: Container of SQLite settings
:param mariadb: Container of MariaDB settings
"""
backend: constants.DatabaseBackend = constants.DatabaseBackend.SQLITE
sqlite: DatabaseSqliteConfig = field(default_factory=DatabaseSqliteConfig.from_env)
mariadb: DatabaseMariaConfig = field(default_factory=DatabaseMariaConfig.from_env)
@classmethod
def from_env(cls):
"""Build dataclass from environment"""
return cls(
backend=constants.DatabaseBackend[
os.environ["SECTION7_DATABASE_BACKEND"].upper()
]
if "SECTION7_DATABASE_BACKEND" in os.environ
else cls.backend,
)
@dataclass
class Section7Config:
"""Global application configuration settings
:param database: Container of database backend settings
:param contact_email: Public administrative contact email for the site
"""
database: DatabaseConfig = field(default_factory=DatabaseConfig.from_env)
contact_email: str = "admin@enp.one"
@classmethod
def from_env(cls):
"""Build dataclass from environment"""
return cls(contact_email=os.getenv("SECTION7_CONTACT_EMAIL", cls.contact_email))
def load() -> Section7Config:
"""Load the application configuration from environment variables
:returns: Populated environment configuration
"""
try:
return Section7Config.from_env()
except (ValueError, TypeError, IndexError, KeyError) as err:
raise RuntimeError("Failed to load application configuration") from err

View File

@ -1,7 +1,17 @@
"""Application constants"""
import enum
class DatabaseBackend(enum.Enum):
"""Enum of supported database backends"""
MARIADB = enum.auto()
SQLITE = enum.auto()
class PayRate(enum.Enum):
"""Enum of payment interval options"""
ANNUALY = enum.auto()
QUARTERLY = enum.auto()
MONTHLY = enum.auto()
@ -12,14 +22,16 @@ class PayRate(enum.Enum):
class Industry(enum.Enum):
pass
"""Enum of industires records can be associated with"""
class Currency(enum.Enum):
pass
"""Enum of currency options"""
class Country(enum.Enum):
"""Enum of country options"""
AF = "Afghanistan"
AX = "Ã…land Islands"
AL = "Albania"

View File

@ -1,15 +1,64 @@
import datetime
import logging
import uuid
from typing import Optional
import peewee
import peewee_plus
from section7 import configuration
from section7 import constants
INTERFACE = peewee.DatabaseProxy()
def initialize(config: Optional[configuration.Section7Config] = None):
"""Initialize the database interface
Defining the database as an
`unconfigured proxy object <http://docs.peewee-orm.com/en/latest/peewee/database.html#setting-the-database-at-run-time>`_
allows it to be configured at runtime based on the config values.
:param config: Populated configuration container object
"""
logger = logging.getLogger(__name__)
config = config or configuration.load()
if config.database.backend == constants.DatabaseBackend.SQLITE:
logger.debug("Using SQLite database backend")
logger.debug(f"Applying SQLite pragmas: {peewee_plus.SQLITE_DEFAULT_PRAGMAS}")
database = peewee.SqliteDatabase(
config.database.sqlite.path, pragmas=peewee_plus.SQLITE_DEFAULT_PRAGMAS
)
elif config.database.backend == constants.DatabaseBackend.MARIADB:
logger.debug("Using MariaDB database backend")
logger.debug(
"Configuring MariaDB:"
f" {config.database.mariadb.username}@{config.database.mariadb.hostname}:{config.database.mariadb.port},"
f" with database '{config.database.mariadb.schema}'"
)
database = peewee.MySQLDatabase(
config.database.mariadb.schema,
host=config.database.mariadb.hostname,
port=config.database.mariadb.port,
user=config.database.mariadb.username,
password=config.database.mariadb.password,
charset="utf8mb4",
)
else:
raise RuntimeError(
f"Invalid storage backend in configuration: {config.database.backend.name}"
)
INTERFACE.initialize(database)
with INTERFACE.atomic():
INTERFACE.create_tables([DisclosureRecord])
class Section7Model(peewee.Model):
"""Base model for defining common fields and attaching database"""