Refactoring to eliminate models. Moved to SQLAlchemy Mapped annotations

README.md

@@ -0,0 +1,168 @@
+![](./src/static/logo.svg)
+
+# Lagomare Gates
+
+A web-based gate access management and control system. Authorized users can remotely open physical car and pedestrian gates via a mobile-friendly PWA. An admin dashboard provides full management of gates, access codes, and users.
+
+## Features
+
+- **Keypass authentication** — users authenticate with an access code; each keypass can have a per-gate allowlist and an optional expiration date
+- **Remote gate control** — integrates with [AVConnect](https://www.avconnect.it) to trigger gate macros
+- **Role-based admin panel** — two roles (`admin`, `manager`) with different permission levels
+- **Access audit log** — every open attempt is logged with timestamp, keypass, gate, IP, and result
+- **Progressive Web App** — installable on mobile devices with offline caching
+
+## Tech Stack
+
+| Layer | Technology |
+|---|---|
+| Backend | FastAPI + Uvicorn |
+| ORM | SQLAlchemy |
+| Database | SQLite |
+| Auth | JWT (HS256) + bcrypt |
+| Credential storage | Fernet symmetric encryption |
+| Gate integration | AVConnect HTTP API |
+| Frontend | Vanilla JS PWA |
+
+## Project Structure
+
+```
+src/
+├── main.py               # App entry point, startup, static file serving
+├── core/
+│   ├── auth.py           # JWT creation/verification, password hashing
+│   ├── config.py         # Settings loaded from environment variables
+│   ├── database.py       # SQLAlchemy models and DB initialization
+│   ├── dependencies.py   # FastAPI dependency injection (auth guards)
+│   └── schemas.py        # Pydantic request/response schemas
+├── models/               # Thin wrappers re-exporting DB models
+├── routers/
+│   ├── auth.py           # POST /api/auth/admin, POST /api/auth/keypass
+│   ├── gates.py          # User-facing gate list and open endpoints
+│   ├── keypasses.py      # Admin keypass CRUD
+│   ├── admins.py         # Admin user management
+│   ├── credentials.py    # AVConnect credential management
+│   └── stats.py          # Access log / statistics
+├── services/
+│   ├── avconnect.py      # AVConnect session management and macro execution
+│   └── gates.py          # Gate open orchestration
+└── static/               # Frontend PWA (index.html, admin.html, JS, CSS)
+data/
+└── gates.db              # SQLite database (auto-created on first run)
+```
+
+## API Endpoints
+
+### Authentication
+
+| Method | Endpoint | Description |
+|---|---|---|
+| POST | `/api/auth/admin` | Admin login — returns JWT |
+| POST | `/api/auth/keypass` | Keypass login — returns JWT |
+
+### User (keypass token required)
+
+| Method | Endpoint | Description |
+|---|---|---|
+| GET | `/api/gates` | List gates accessible to the authenticated keypass |
+| POST | `/api/gates/{gate_id}/open` | Open a gate |
+
+### Admin — Gates (manager+)
+
+| Method | Endpoint | Description |
+|---|---|---|
+| GET | `/api/admin/gates` | List all gates |
+| POST | `/api/admin/gates` | Create a gate *(admin only)* |
+| PUT | `/api/admin/gates/{gate_id}` | Update a gate *(admin only)* |
+| DELETE | `/api/admin/gates/{gate_id}` | Delete a gate *(admin only)* |
+| POST | `/api/admin/gates/{gate_id}/open` | Manually open a gate |
+
+### Admin — Keypasses (manager+)
+
+| Method | Endpoint | Description |
+|---|---|---|
+| GET | `/api/admin/keypasses` | List all keypasses |
+| POST | `/api/admin/keypasses` | Create a keypass |
+| PATCH | `/api/admin/keypasses/{kp_id}` | Update a keypass |
+| DELETE | `/api/admin/keypasses/{kp_id}` | Revoke a keypass |
+
+### Admin — Users (admin only)
+
+| Method | Endpoint | Description |
+|---|---|---|
+| GET | `/api/admin/admins` | List admin users |
+| POST | `/api/admin/admins` | Create an admin user |
+| DELETE | `/api/admin/admins/{username}` | Delete an admin user |
+| PATCH | `/api/admin/admins/{username}/password` | Change password |
+
+### Admin — AVConnect Credentials (admin only)
+
+| Method | Endpoint | Description |
+|---|---|---|
+| GET | `/api/admin/credentials` | View stored credentials |
+| PUT | `/api/admin/credentials` | Create or update credentials |
+
+### Admin — Statistics (manager+)
+
+| Method | Endpoint | Description |
+|---|---|---|
+| GET | `/api/admin/stats` | Retrieve the last 500 access log entries |
+
+## Configuration
+
+All settings are read from environment variables.
+
+| Variable | Default | Description |
+|---|---|---|
+| `SECRET_KEY` | Random 32 bytes | JWT signing key and Fernet encryption key. **Set this explicitly in production.** |
+| `ADMIN_USERNAME` | `admin` | Username for the initial admin account |
+| `ADMIN_PASSWORD` | *(none)* | Password for the initial admin account. If unset, no seed account is created. |
+| `APP_PORT` | `8000` | HTTP port the server listens on |
+| `DATABASE_URL` | `sqlite:///data/gates.db` | SQLAlchemy database URL |
+| `MOCK_AVCONNECT` | `false` | Set to `true` to skip real AVConnect calls (always returns success — useful for development) |
+
+## Running with Docker Compose
+
+```bash
+# Copy and edit the compose file to set your SECRET_KEY
+docker compose up -d
+```
+
+The default `docker-compose.yml` starts the service on port `8000` with the initial admin credentials `admin` / `changeme`. Change `ADMIN_PASSWORD` and set a strong `SECRET_KEY` before deploying.
+
+The `./data` directory is mounted into the container so the SQLite database persists across restarts.
+
+## Running Locally
+
+```bash
+python -m venv venv
+source venv/bin/activate
+pip install -r requirements.txt
+
+export SECRET_KEY="change-me"
+export ADMIN_USERNAME="admin"
+export ADMIN_PASSWORD="changeme"
+
+uvicorn src.main:app --reload --port 8000
+```
+
+The application is then available at:
+
+- **User interface** — `http://localhost:8000/`
+- **Admin dashboard** — `http://localhost:8000/admin`
+
+## AVConnect Integration
+
+Gates are controlled through the AVConnect platform. Each gate is mapped to an AVConnect *macro ID*. When a gate open request is received, the service:
+
+1. Authenticates with AVConnect using the stored credentials (session is cached in the database)
+2. Executes the configured macro for the gate
+
+Credentials (password) are stored encrypted in the database using Fernet symmetric encryption derived from `SECRET_KEY`.
+
+## Roles
+
+| Role | Permissions |
+|---|---|
+| `admin` | Full access — all endpoints including gate/user/credential management |
+| `manager` | Gate open, keypass management, statistics — cannot manage admin users, AVConnect credentials, or create/delete gates |

src/core/database.py

@@ -1,8 +1,9 @@
 import os
 from datetime import datetime
+from typing import Optional
 
-from sqlalchemy import Boolean, Column, DateTime, Integer, String, Text, create_engine
+from sqlalchemy import Boolean, String, Text, create_engine
-from sqlalchemy.orm import DeclarativeBase, sessionmaker
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker
 
 _HERE = os.path.dirname(os.path.abspath(__file__))
 _SRC_DIR = os.path.dirname(_HERE)
@@ -25,57 +26,57 @@ class Base(DeclarativeBase):
 class GateDB(Base):
     __tablename__ = "gates"
 
-    id = Column(Integer, primary_key=True, autoincrement=True)
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
-    name = Column(String, nullable=False)
+    name: Mapped[str] = mapped_column(String, nullable=False)
-    gate_type = Column(String, nullable=False)        # 'car' | 'pedestrian'
+    gate_type: Mapped[str] = mapped_column(String, nullable=False)        # 'car' | 'pedestrian'
-    avconnect_macro_id = Column(String, nullable=False)  # AVConnect macro ID
+    avconnect_macro_id: Mapped[str] = mapped_column(String, nullable=False)  # AVConnect macro ID
-    status = Column(String, default="enabled")        # 'enabled' | 'disabled'
+    status: Mapped[str] = mapped_column(String, default="enabled")        # 'enabled' | 'disabled'
 
 
 class ApiCredential(Base):
     __tablename__ = "api_credentials"
 
-    id = Column(Integer, primary_key=True, autoincrement=True)
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
-    username = Column(String, nullable=False)
+    username: Mapped[str] = mapped_column(String, nullable=False)
-    password_enc = Column(String, nullable=False)   # Fernet-encrypted
+    password_enc: Mapped[str] = mapped_column(String, nullable=False)   # Fernet-encrypted
-    session_id = Column(String, nullable=True)
+    session_id: Mapped[Optional[str]] = mapped_column(String, nullable=True)
 
 
 class Keypass(Base):
     __tablename__ = "keypasses"
 
-    id = Column(Integer, primary_key=True, autoincrement=True)
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
-    code = Column(String, unique=True, nullable=False)
+    code: Mapped[str] = mapped_column(String, unique=True, nullable=False)
-    description = Column(Text, nullable=False)
+    description: Mapped[str] = mapped_column(Text, nullable=False)
-    created_at = Column(DateTime, nullable=False)
+    created_at: Mapped[datetime] = mapped_column(nullable=False)
-    expires_at = Column(DateTime, nullable=True)     # NULL = never expires
+    expires_at: Mapped[Optional[datetime]] = mapped_column(nullable=True)     # NULL = never expires
-    revoked = Column(Boolean, default=False)
+    revoked: Mapped[bool] = mapped_column(Boolean, default=False)
-    revoked_at = Column(DateTime, nullable=True)
+    revoked_at: Mapped[Optional[datetime]] = mapped_column(nullable=True)
-    allowed_gates = Column(Text, nullable=True)      # JSON list of gate IDs; NULL = all gates
+    allowed_gates: Mapped[Optional[str]] = mapped_column(Text, nullable=True)  # JSON list of gate IDs; NULL = all gates
 
 
 class GateAccessLog(Base):
     __tablename__ = "gate_access_logs"
 
-    id = Column(Integer, primary_key=True, autoincrement=True)
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
-    timestamp = Column(DateTime, nullable=False)
+    timestamp: Mapped[datetime] = mapped_column(nullable=False)
-    keypass_id = Column(Integer, nullable=False)
+    keypass_id: Mapped[int] = mapped_column(nullable=False)
-    keypass_code = Column(String, nullable=False)
+    keypass_code: Mapped[str] = mapped_column(String, nullable=False)
-    gate_id = Column(Integer, nullable=False)
+    gate_id: Mapped[int] = mapped_column(nullable=False)
-    gate_name = Column(String, nullable=False)
+    gate_name: Mapped[str] = mapped_column(String, nullable=False)
-    ip_address = Column(String, nullable=True)
+    ip_address: Mapped[Optional[str]] = mapped_column(String, nullable=True)
-    user_agent = Column(Text, nullable=True)
+    user_agent: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
-    success = Column(Boolean, nullable=False)
+    success: Mapped[bool] = mapped_column(Boolean, nullable=False)
-    error = Column(Text, nullable=True)
+    error: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
 
 
 class AdminUser(Base):
     __tablename__ = "admin_users"
 
-    id = Column(Integer, primary_key=True, autoincrement=True)
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
-    username = Column(String, unique=True, nullable=False)
+    username: Mapped[str] = mapped_column(String, unique=True, nullable=False)
-    password_hash = Column(String, nullable=False)
+    password_hash: Mapped[str] = mapped_column(String, nullable=False)
-    role = Column(String, nullable=False, default="admin")  # 'admin' | 'manager'
+    role: Mapped[str] = mapped_column(String, nullable=False, default="admin")  # 'admin' | 'manager'
 
 
 def get_db():

src/models/__init__.py

@@ -1,9 +0,0 @@
-from .credential import Credential
-from .gate import Gate
-from .status import Status
-
-__all__ = [
-    "Credential",
-    "Gate",
-    "Status"
-]

src/models/credential.py

@@ -1,5 +0,0 @@
-class Credential:
-    def __init__(self, username: str, password: str):
-        self.username = username
-        self.password = password
-        self.sessionid = None

src/models/gate.py

@@ -1,9 +0,0 @@
-from .status import Status
-from .credential import Credential
-
-class Gate:
-    def __init__(self, id: str, name: str, status: Status = Status.ENABLED):
-        self.id = id
-        self.name = name
-        self.status = status if isinstance(status, Status) else Status(status)
-

src/models/status.py

@@ -1,5 +0,0 @@
-from enum import Enum
-
-class Status(Enum):
-    ENABLED = 1
-    DISABLED = 0

src/routers/gates.py

@@ -8,7 +8,6 @@ from sqlalchemy.orm import Session
 from core.auth import decrypt_secret
 from core.database import ApiCredential, GateAccessLog, GateDB, Keypass, get_db
 from core.dependencies import require_admin, require_manager, require_keypass
-from models import Credential, Gate as GateModel, Status
 from core.schemas import GateCreate, GateResponse
 from services.gates import call_open_gate
 
@@ -86,17 +85,15 @@ async def admin_open_gate(
     if not cred_db:
         raise HTTPException(503, "AVConnect credentials not configured")
 
-    credential = Credential(
-        username=cred_db.username,
-        password=decrypt_secret(cred_db.password_enc),
-    )
-    credential.sessionid = cred_db.session_id
-
-    gate = GateModel(id=gate_db.avconnect_macro_id, name=gate_db.name, status=Status.ENABLED)
     ip = request.headers.get("X-Forwarded-For", request.client.host if request.client else None)
     ua = request.headers.get("User-Agent")
 
-    success, error_msg, new_sid = call_open_gate(gate, credential)
+    success, error_msg, new_sid = call_open_gate(
+        gate_db.avconnect_macro_id,
+        cred_db.username,
+        decrypt_secret(cred_db.password_enc),
+        cred_db.session_id,
+    )
 
     db.add(GateAccessLog(
         timestamp=datetime.utcnow(),
@@ -150,14 +147,6 @@ async def open_gate(
     if not cred_db:
         raise HTTPException(503, "AVConnect credentials not configured")
 
-    credential = Credential(
-        username=cred_db.username,
-        password=decrypt_secret(cred_db.password_enc),
-    )
-    credential.sessionid = cred_db.session_id
-
-    gate_status = Status.ENABLED if gate_db.status == "enabled" else Status.DISABLED
-    gate = GateModel(id=gate_db.avconnect_macro_id, name=gate_db.name, status=gate_status)
     ip = request.headers.get("X-Forwarded-For", request.client.host if request.client else None)
     ua = request.headers.get("User-Agent")
 
@@ -165,7 +154,12 @@ async def open_gate(
     if allowed is not None and gate_id not in allowed:
         raise HTTPException(403, "This keypass does not have access to this gate")
 
-    success, error_msg, new_sid = call_open_gate(gate, credential)
+    success, error_msg, new_sid = call_open_gate(
+        gate_db.avconnect_macro_id,
+        cred_db.username,
+        decrypt_secret(cred_db.password_enc),
+        cred_db.session_id,
+    )
 
     db.add(GateAccessLog(
         timestamp=datetime.utcnow(),

src/services/__init__.py

@@ -1,9 +1,7 @@
 from .avconnect import AVConnectAPI
-from .gates import GatesService, OpenResult, call_open_gate
+from .gates import call_open_gate
 
 __all__ = [
     "AVConnectAPI",
-    "GatesService",
-    "OpenResult",
     "call_open_gate",
 ]

src/services/avconnect.py

@@ -1,18 +1,18 @@
 import requests
 from fake_useragent import UserAgent
-from models import Credential
 
 class AVConnectAPI:
     _BASE_URL = "https://www.avconnect.it"
 
-    def __init__(self, credentials: Credential):
+    def __init__(self, username: str, password: str, session_id: str | None = None):
         self._ua = UserAgent(browsers=["Chrome Mobile"], os=["Android"], platforms=["mobile"]).random
-        self._credentials = credentials
+        self._username = username
+        self._password = password
         self._session = requests.Session()
         self._authenticated = False
 
-        if credentials.sessionid:
+        if session_id:
-            self._session.cookies.set("PHPSESSID", credentials.sessionid)
+            self._session.cookies.set("PHPSESSID", session_id)
             self._authenticated = True
 
     def _authenticate(self) -> bool:
@@ -21,7 +21,7 @@ class AVConnectAPI:
             "User-Agent": self._ua,
             "Content-Type": "application/x-www-form-urlencoded"
         }
-        payload = f"userid={self._credentials.username}&password={self._credentials.password}&entra=Login"
+        payload = f"userid={self._username}&password={self._password}&entra=Login"
         response = self._session.post(login_url, data=payload, headers=headers)
         if response.ok and "PHPSESSID" in self._session.cookies:
             self._authenticated = True
@@ -30,7 +30,7 @@ class AVConnectAPI:
         return False
     
     def _check_sessionid(self) -> bool:
-        if not self._authenticated or not self._credentials.sessionid:
+        if not self._authenticated or not self._session.cookies.get("PHPSESSID"):
             return False
         exec_url = f"{self._BASE_URL}/exemacrocom.php"
         headers = {

src/services/gates.py

@@ -1,38 +1,26 @@
-from dataclasses import dataclass
 from typing import Optional
 
-from models import Credential, Status, Gate
 from .avconnect import AVConnectAPI
 
 
-@dataclass
+def call_open_gate(
-class OpenResult:
+    macro_id: str,
-    success: bool
+    username: str,
-    error: Optional[str] = None
+    password: str,
-    new_session_id: Optional[str] = None
+    session_id: Optional[str] = None,
-
+) -> tuple[bool, Optional[str], Optional[str]]:
-
-class GatesService:
-    def open_gate(self, gate: Gate, credentials: Credential) -> OpenResult:
-        if gate.status == Status.DISABLED:
-            return OpenResult(success=False, error="Gate is disabled")
-        try:
-            api = AVConnectAPI(credentials)
-            ok = api.exec_gate_macro(gate.id)
-            new_sid = api._session.cookies.get("PHPSESSID")
-            if not ok:
-                return OpenResult(success=False, error="Gate did not confirm open", new_session_id=new_sid)
-            return OpenResult(success=True, new_session_id=new_sid)
-        except Exception as e:
-            return OpenResult(success=False, error=str(e))
-
-
-def call_open_gate(gate: Gate, credentials: Credential) -> tuple[bool, Optional[str], Optional[str]]:
     """Attempt to open a gate. Returns (success, error_msg, new_session_id).
     Respects the MOCK_AVCONNECT environment variable.
     """
     from core.config import MOCK_AVCONNECT
     if MOCK_AVCONNECT:
         return True, None, None
-    result = GatesService().open_gate(gate, credentials)
+    try:
-    return result.success, result.error, result.new_session_id
+        api = AVConnectAPI(username, password, session_id)
+        ok = api.exec_gate_macro(macro_id)
+        new_sid = api._session.cookies.get("PHPSESSID")
+        if not ok:
+            return False, "Gate did not confirm open", new_sid
+        return True, None, new_sid
+    except Exception as e:
+        return False, str(e), None