[settings] fuck it, rework the settings again

this time using pydantic-settings as a base
- removed all os.getenv calls
- removed the secret_key default option
- reworked database loading, creating tables
- prob. something else also but its 4:30 and i have to sleep
This commit is contained in:
2026-07-31 04:33:18 +02:00
parent bad4e8dd1b
commit 158b430235
11 changed files with 100 additions and 89 deletions

View File

@@ -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)

View File

@@ -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)

View File

@@ -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

View File

@@ -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):

View File

@@ -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

View File

@@ -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

View File

@@ -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()

View File

@@ -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]

View File

@@ -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():

View File

@@ -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()

2
uv.lock generated
View File

@@ -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" },