diff --git a/app/__init__.py b/app/__init__.py index e4c3274..e69de29 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -1,7 +0,0 @@ -from fastapi import FastAPI - -from .controllers import cardManager, userManager - -app = FastAPI() -app.include_router(userManager.user_router) -app.include_router(cardManager.card_router) diff --git a/app/main.py b/app/main.py index 7910363..210c59b 100644 --- a/app/main.py +++ b/app/main.py @@ -1,15 +1,10 @@ import logging -import os from contextlib import asynccontextmanager -from dotenv import load_dotenv from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from fastapi.security import OAuth2PasswordBearer -load_dotenv() - -# ruff: disable[E402] from app.controllers import ( aaManager, cardManager, @@ -22,9 +17,7 @@ from app.services.auth import create_first_user, token_router from app.services.database import create_db_and_tables, get_db_session from app.services.door import DoorController, init_controller from app.services.scanner import BackgroundScanner -from app.services.settings import verify_settings - -# ruff: enable[E402] +from app.services.settings import settings logger = logging.getLogger(__name__) @@ -35,17 +28,16 @@ logging.basicConfig(level=logging.INFO) @asynccontextmanager async def lifespan(app: FastAPI): - verify_settings() create_db_and_tables() create_first_user(db=get_db_session()) init_controller( DoorController( - lock_pin=os.getenv("LOCK_PIN", 17), - unlock_pin=os.getenv("UNLOCK_PIN", 18), - mock_factory=os.getenv("MOCK_GPIO", True), + lock_pin=settings.lock_pin, + unlock_pin=settings.unlock_pin, + mock_factory=settings.mock_gpio, ), ) - if not os.getenv("DISABLE_CARDS"): + if not settings.disable_cards: scanner.start() logger.info("-" * 63) diff --git a/app/services/auth.py b/app/services/auth.py index 5b54232..d4392db 100644 --- a/app/services/auth.py +++ b/app/services/auth.py @@ -1,5 +1,4 @@ import logging -import os import secrets import string from datetime import UTC, datetime, timedelta @@ -14,10 +13,11 @@ from sqlmodel import Session, select from app.model.models import Token, TokenData, UserDB from app.services.database import add_and_refresh, get_session +from app.services.settings import settings logger = logging.getLogger(__name__) -SECRET_KEY = os.getenv("SECRET_KEY", default="ff" * 16) +SECRET_KEY = settings.secret_key ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 120 diff --git a/app/services/database.py b/app/services/database.py index f1adadc..7b19034 100644 --- a/app/services/database.py +++ b/app/services/database.py @@ -1,38 +1,48 @@ import logging -from os import getenv, path +from functools import lru_cache +from sqlalchemy import inspect from sqlmodel import Session, SQLModel, create_engine +from app.services.settings import settings + logger = logging.getLogger(__name__) -SQLALCHEMY_DATABASE_PATH = getenv("SQLALCHEMY_DATABASE_PATH", "./gatekeeper.db") -SQLALCHEMY_DATABASE_URL = "sqlite:///" + SQLALCHEMY_DATABASE_PATH -engine = create_engine(SQLALCHEMY_DATABASE_URL) +@lru_cache +def get_engine(): + return create_engine( + settings.sqlalchemy_database_url, connect_args={"check_same_thread": False} + ) def create_db_and_tables(): - if not path.exists(SQLALCHEMY_DATABASE_PATH): - SQLModel.metadata.create_all(engine) + inspector = inspect(get_engine()) + existing_tables = inspector.get_table_names() + if not existing_tables: + SQLModel.metadata.create_all(get_engine()) from alembic.config import Config from alembic import command - alembic_cfg = Config("./alembic.ini") - alembic_cfg.attributes["sqlalchemy.url"] = SQLALCHEMY_DATABASE_URL + alembic_cfg = Config(settings.alembic_config) + alembic_cfg.set_main_option("sqlalchemy.url", str(get_engine().url)) + alembic_cfg.attributes["sqlalchemy.url"] = settings.sqlalchemy_database_url command.stamp(alembic_cfg, "head") logger.info("Database created and tables initialized.") else: - logger.info("Database already exists") + logger.info( + "Database already initialized (%d tables found).", len(existing_tables) + ) def get_session(): - with Session(engine) as db: + with Session(get_engine()) as db: yield db def get_db_session(): - return Session(engine) + return Session(get_engine()) def add_and_refresh(db: Session, obj): diff --git a/app/services/door.py b/app/services/door.py index fcc0d4d..94339c8 100644 --- a/app/services/door.py +++ b/app/services/door.py @@ -18,8 +18,8 @@ logger = logging.getLogger(__name__) class DoorController: def __init__( self, - lock_pin: str = 17, # connected to 20 on the esp - unlock_pin: str = 18, # connected to 21 on the esp + lock_pin: int = 17, # connected to 20 on the esp + unlock_pin: int = 18, # connected to 21 on the esp mock_factory: bool = False, ): self._is_open: bool = False diff --git a/app/services/scanner.py b/app/services/scanner.py index 72e056c..b5ef31f 100644 --- a/app/services/scanner.py +++ b/app/services/scanner.py @@ -26,14 +26,15 @@ from smartcard.CardType import AnyCardType from smartcard.Exceptions import CardRequestTimeoutException from app.services.door import checkAccess, openDoor +from app.services.settings import settings logger = logging.getLogger(__name__) # ENV vars load_dotenv() -MIFARE_APP_MASTER_KEY = os.getenv("MIFARE_APP_MASTER_KEY") -MIFARE_ACL_READ_BASE_KEY = os.getenv("MIFARE_ACL_READ_BASE_KEY") -MIFARE_ACL_WRITE_BASE_KEY = os.getenv("MIFARE_ACL_WRITE_BASE_KEY") +MIFARE_APP_MASTER_KEY = settings.mifare_app_master_key +MIFARE_ACL_READ_BASE_KEY = settings.mifare_acl_read_base_key +MIFARE_ACL_WRITE_BASE_KEY = settings.mifare_acl_write_base_key # Constants MIFARE_APP_ID = "DEAFFE" # 7 bytes diff --git a/app/services/settings.py b/app/services/settings.py index 5f84f88..e22c766 100644 --- a/app/services/settings.py +++ b/app/services/settings.py @@ -1,52 +1,70 @@ import logging -import os from functools import lru_cache +from pydantic_settings import BaseSettings, SettingsConfigDict + logger = logging.getLogger(__name__) +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_file=".env") + + secret_key: str + sqlalchemy_database_url: str = "sqlite:///./gatekeeper.db" + lock_pin: int = 17 + unlock_pin: int = 18 + mock_gpio: bool = True + alembic_config: str = "./alembic.ini" + + disable_cards: bool = False + mifare_app_master_key: str | None = None + mifare_acl_read_base_key: str | None = None + mifare_acl_write_base_key: str | None = None + + @lru_cache(1) -def disable_cards_check(): - if os.environ.get("DISABLE_CARDS", "").lower() in ("1", "true"): - return True - - for env in ( - "MIFARE_APP_MASTER_KEY", - "MIFARE_ACL_READ_BASE_KEY", - "MIFARE_ACL_WRITE_BASE_KEY", - ): - if not os.getenv(env): +def _create_settings(): + settings = Settings() + if not settings.disable_cards: + missing = [ + name.upper() + for name in ( + "mifare_app_master_key", + "mifare_acl_read_base_key", + "mifare_acl_write_base_key", + ) + if not getattr(settings, name) + ] + if missing: logger.critical( - f"Missing environment variable for scanner start: {env} \n Card scanner and related funcionality is disabled!" # noqa: E501 + "Missing environment variable for scanner start: %s" + "Card scanner and related funcionality is disabled!", + ", ".join(missing), ) - return True - return False + settings.disable_cards = True + return settings -disable_cards = disable_cards_check() +class _SettingsProxy: + _instance: Settings | None = None -if not disable_cards: - mifare_app_master_key = os.getenv("MIFARE_APP_MASTER_KEY") - mifare_acl_read_base_key = os.getenv("MIFARE_ACL_READ_BASE_KEY") - mifare_acl_write_base_key = os.getenv("MIFARE_ACL_WRITE_BASE_KEY") + def _load(self) -> Settings: + if self._instance is None: + self._instance = _create_settings() + return self._instance + + def __getattr__(self, name: str): + return getattr(self._load(), name) + + def __setattr__(self, name: str, value): + if name == "_instance": + super().__setattr__(name, value) + else: + setattr(self._load(), name, value) + + def reset(self) -> None: + self._instance = None + _create_settings.cache_clear() -def verify_settings(): - important_envs = ["SECRET_KEY"] - other_envs = ["SQLALCHEMY_DATABASE_PATH", "LOCK_PIN", "UNLOCK_PIN", "MOCK_GPIO"] - for setting in important_envs: - if not os.getenv(setting): - raise ValueError( - f"Missing critical environment variable {setting}. Stopping..." - ) - for setting in other_envs: - if setting not in os.environ: - logger.warning(f"Env var {setting} not set. Continuing with defaults.") - - -verify_settings() -secret_key = os.getenv("SECRET_KEY") -sqlalchemy_database_path = os.getenv("SQLALCHEMY_DATABASE_PATH") -lock_pin = os.getenv("LOCK_PIN") -unlock_pin = os.getenv("UNLOCK_PIN") -mock_gpio = os.getenv("MOCK_GPIO") +settings = _SettingsProxy() diff --git a/pyproject.toml b/pyproject.toml index 92ba7c0..b26351e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,6 +19,7 @@ dependencies = [ "alembic>=1.18.5", "ruff>=0.16.0", "gpiozero>=2.0.1.post3", + "pydantic-settings>=2.13.1", ] [tool.uv.sources] diff --git a/test/conftest.py b/test/conftest.py index 92ce9f9..8f12ee3 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,3 +1,4 @@ +import os from datetime import time import pytest @@ -5,6 +6,9 @@ from fastapi.testclient import TestClient from sqlalchemy.pool import StaticPool from sqlmodel import Session, SQLModel, create_engine +os.environ["SECRET_KEY"] = "ff" * 16 + +from app.main import app from app.model.models import ( AccessAuthorizationDB, Card, @@ -24,16 +28,6 @@ engine = create_engine( ) -@pytest.fixture -def app(): - import os - - os.environ["SECRET_KEY"] = "ff" * 16 - from app.main import app - - return app - - @pytest.fixture(scope="function") def db_session(): """Create a fresh database session for each test.""" @@ -44,7 +38,7 @@ def db_session(): @pytest.fixture(scope="function") -def client(app, db_session): +def client(db_session): """Create a test client with a database session override.""" def override_get_session(): diff --git a/test/test_services/test_database.py b/test/test_services/test_database.py index 5030922..2b55570 100644 --- a/test/test_services/test_database.py +++ b/test/test_services/test_database.py @@ -9,10 +9,10 @@ def test_create_db_and_tables(): # This is primarily an integration test from sqlalchemy import inspect - from app.services.database import engine + from app.services.database import get_engine create_db_and_tables() - inspector = inspect(engine) + inspector = inspect(get_engine()) # Check that tables exist tables = inspector.get_table_names() diff --git a/uv.lock b/uv.lock index 7449981..85457fa 100644 --- a/uv.lock +++ b/uv.lock @@ -639,6 +639,7 @@ dependencies = [ { name = "gpiozero" }, { name = "poetry" }, { name = "pwdlib", extra = ["argon2"] }, + { name = "pydantic-settings" }, { name = "pyjwt", extra = ["crypto"] }, { name = "pyscard" }, { name = "pytest" }, @@ -657,6 +658,7 @@ requires-dist = [ { name = "gpiozero", specifier = ">=2.0.1.post3" }, { name = "poetry", specifier = ">=2.3.4" }, { name = "pwdlib", extras = ["argon2"], specifier = ">=0.3.0" }, + { name = "pydantic-settings", specifier = ">=2.13.1" }, { name = "pyjwt", extras = ["crypto"], specifier = ">=2.12.1" }, { name = "pyscard", specifier = ">=2.3.1" }, { name = "pytest", specifier = ">=9.0.3" },