utp_service/CLAUDE.md

UTP Service — Project Reference

Fleet management web service for parking/access control equipment (CAPS system).

Stack

Layer	Tech
Backend	FastAPI + SQLAlchemy 2.0 async (asyncpg), Alembic, asyncssh, psycopg3
Frontend	React 18 + TypeScript, Vite, Ant Design 5, TanStack React Query 5, Zustand
DB	PostgreSQL 16
Infra	Docker Compose

Run: docker compose up — backend :8005, frontend :3005, postgres :5432

Project Layout

utp_service/
├── backend/
│   ├── app/
│   │   ├── main.py              # FastAPI app, routers, lifespan
│   │   ├── config.py            # Settings (DATABASE_URL, SECRET_KEY, ENCRYPTION_KEY)
│   │   ├── dependencies.py      # get_db, get_current_user
│   │   ├── models/              # SQLAlchemy ORM models
│   │   ├── routers/             # API route handlers
│   │   ├── schemas/             # Pydantic request/response schemas
│   │   ├── services/            # Business logic
│   │   ├── plugins/             # Plugin registry + builtin plugins
│   │   ├── workers/             # Background workers (monitor, startup)
│   │   └── middleware/auth.py   # Auth middleware
│   └── alembic/versions/        # Numbered migrations 0001–0009
└── frontend/
    └── src/
        ├── App.tsx              # React Router routes
        ├── pages/               # Page-level components
        ├── api/                 # React Query hooks + API calls
        ├── hooks/               # useSSE, useMonitorSSE
        ├── components/          # AppLayout, StatusBadge
        └── store/auth.ts        # Zustand auth store

Database Session Pattern

autoflush=False — must call await db.flush() explicitly after db.add() before any SELECT that needs to see the new rows. session.begin() auto-commits on handler return.

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        async with session.begin():
            yield session

Models

Object (objects table)

Represents a physical site/parking lot.

Key fields: name, description, city, client, server_ip (main CAPS server IP), db_host/port/name/user/db_pass_enc/db_table/db_via_tunnel (CAPS PostgreSQL connection), ssh_user/ssh_pass_enc/ssh_port (SSH credentials, shared across site), is_active, tags (M2M → Tag)

Device (devices table)

Network device at a site.

Key fields: object_id (FK), ip (unique per object: uq_device_object_ip), hostname, mac, category, role, location, vendor, model, source (manual/db_sync/auto_ssh/csv/sheets), status (unknown/online/offline/degraded), monitor_enabled, rack_id (FK → Rack), connection_params (JSONB), device_meta (JSONB — e.g. {"plugin": "camera_1", "cls": "..."}), capabilities (ARRAY), ssh_user_override/ssh_pass_override_enc/ssh_port_override

Device categories: main_server, vm, router, embedded, camera, io_board, bank_terminal, cash_register, other Device roles: plate, face, overview, other Device locations: server_room, rack, street

Rack (racks table)

id, object_id, name, description, location, zone_id (FK → Zone)

Zone (zones table)

id, object_id, name, description, racks relationship

Task (tasks table)

id, type, status (pending/running/success/partial/failed/stale/cancelled), target_scope (JSONB — see Task System), params (JSONB), result (JSONB), created_by, timestamps, → TaskEvent, TaskResult relationships

Alert (alerts table)

id, device_id, status (open/acknowledged/resolved), message, timestamps

ConfigSnapshot / SnapshotContent

Content-addressed: SnapshotContent(sha256 PK, content), ConfigSnapshot(device_id, sha256 FK, path, captured_at)

Tag (tags table)

id, name, color, tag_type; M2M with objects via object_tags

Routers (all under /api/v1/)

Router file	Prefix	Key endpoints
auth.py	/auth	POST login, POST refresh, POST logout
objects.py	/objects	CRUD + GET /meta (cities/clients) + POST /{id}/discover + POST /{id}/sync
devices.py	/objects/{id}/devices	CRUD + POST /import + POST /import/preview
global_devices.py	/devices	GET all devices across objects (filters: category, city, client, object_id, status)
racks.py	/objects/{id}/racks	CRUD
zones.py	/objects/{id}/zones	CRUD
tasks.py	/tasks	POST create + GET list/detail + GET /{id}/events (SSE)
events.py	/events	GET SSE stream (task_id param)
alerts.py	/alerts	GET list + POST /{id}/acknowledge + POST /{id}/resolve
snapshots.py	/snapshots	GET list + GET /{id}/diff
tags.py	/tags	CRUD
admin.py	/admin	User mgmt + plugin control + health

Services

File	Purpose
device_discovery.py	SSH-based discovery: SSH → rack PC → parse /etc/caps/config.ini → extract plugin IPs → upsert devices
object_db.py	Sync devices from CAPS PostgreSQL DB
task.py	Task dispatch, per-device async execution (concurrency: 20)
ssh.py	asyncssh client with concurrency limits
crypto.py	Fernet encrypt/decrypt for passwords (db_pass_enc, ssh_pass_enc)
sse.py	SSEManager pub/sub for Server-Sent Events
camera.py	Hikvision ISAPI HTTP client
snapshot.py	SHA-256 content-addressed config snapshot storage
sheets.py	Google Sheets device import (gspread)
auth.py	JWT + bcrypt, access (15min) + refresh (7d) tokens

SSH Discovery (device_discovery.py)

Flow:

1. Resolve CAPS DB connection: use explicit obj.db_* fields OR autodetect by SSHing into obj.server_ip → read /etc/caps/config.ini → extract db_uri
2. Query racks table → get (external_id, rack_host_ip, rack_name) tuples
3. For each rack: SSH into rack_host_ip → read /etc/caps/config.ini
4. Parse config with RawConfigParser(strict=False) (strict=False needed for duplicate keys like ini= in uwsgi)
5. Extract devices from [plugin:*] sections by parsing URI fields (tcp://, http://) — skip serial://, localhost
6. Upsert: SELECT by (object_id, ip), INSERT or UPDATE category/role/hostname
7. await db.flush() after each rack (autoflush=False — without flush, SELECT won't see pending db.add() rows)

Hostname format: rack{rack_name}-{plugin_name} e.g. rack01-camera_grz

Task System

target_scope JSONB shapes:

• {"type": "device", "id": 123}
• {"type": "object", "id": 4}
• {"type": "category", "object_id": 4, "category": "main_server"}
• {"type": "device_list", "ids": [1, 2, 3]}

Task events streamed via SSE at GET /api/v1/events?task_id={id} Monitor events at GET /api/v1/events?task_id=__monitor__

SSE fix: onerror must throw err to prevent fetchEventSource auto-retry loop on 401.

Frontend Routes

/                              → redirect to /inventory/objects
/inventory/objects             → ObjectsPage (mode=inventory) — city-grouped list
/inventory/objects/:id         → InventoryObjectDetailPage — device CRUD, SSH discovery
/inventory/objects/:id/snapshots → SnapshotsPage
/work/objects                  → ObjectsPage (mode=work) — city-grouped list
/work/objects/:id              → WorkObjectDetailPage — task execution, device status
/work/devices                  → WorkDevicesPage — global device list with filters

Layout nav: ИНВЕНТАРЬ (DatabaseOutlined) | РАБОТА (ControlOutlined, AppstoreOutlined)

Frontend API Hooks (src/api/)

• objects.ts: useObjects, useObject, useCreateObject, useUpdateObject, useDeleteObject, useObjectsMeta (cities/clients autocomplete), useDiscoverObject, useSyncFromDB, useSyncFromSheets
• devices.ts: useDevices, useDevice, useCreateDevice, useUpdateDevice, useDeleteDevice, useImportDevicesPreview, useImportDevices, useAllDevices (global)
• tasks.ts: useTasks, useTaskDetail, useCreateTask, useTaskSSE
• alerts.ts: useAlerts, useAcknowledgeAlert, useResolveAlert
• snapshots.ts: useSnapshots, useSnapshotDetail, useSnapshotDiff, useCaptureSnapshot

Plugins (builtin)

Located in backend/app/plugins/builtin/:

• camera/: get_status (Hikvision ISAPI), set_ntp
• mikrotik/: get_config, get_ntp, set_ntp (RouterOS via SSH)
• ssh/: debian_system_info, get_file, get_timedatectl, ssh_command
• common/: ping

Migrations

Apply via: docker compose exec backend alembic upgrade head

Files in backend/alembic/versions/:

• 0001: initial schema
• 0002: tasks
• 0003: racks
• 0004: zones
• 0005: snapshots
• 0006: alerts
• 0007: plugins
• 0008: inventory (vendor, location, connection_params on devices)
• 0009: city, client on objects

Known Gotchas

• autoflush=False: always await db.flush() after db.add() before subsequent SELECTs in the same transaction
• await db.refresh(obj) — no attribute_names arg, otherwise server-generated columns (updated_at) aren't refreshed
• RawConfigParser(strict=False) — CAPS configs have duplicate keys in [uwsgi] section
• SSH discovery: onerror(err) { throw err } in fetchEventSource to stop auto-retry on non-200
• Encrypted fields stored as Fernet ciphertext; use crypto.encrypt() / crypto.decrypt()