Lukas
52a4d04d92
Convert inconsistent modeling style to standardized Pydantic models for all external-facing data structures while maintaining TypedDict compatibility where appropriate for internal layer-private structures. Changes: - Converted IpLookupResult TypedDict to use IpLookupResponse Pydantic model in jail_service.lookup_ip() for consistency with routers - Added GeoCacheEntry Pydantic model for geo cache repository rows - Converted GeoCacheRow TypedDict to use GeoCacheEntry alias - Converted ImportLogRow TypedDict to use ImportLogEntry alias - Updated routers and services to work with Pydantic models - Updated all tests to use Pydantic model field access (attributes) instead of dict subscripting Documentation: - Added 'Model Type Usage by Layer' section to Backend-Development.md - Defines when TypedDict is allowed (internal structures) vs Pydantic (external-facing, cross-boundary data) - Provides clear guidance on modeling conventions per layer Benefits: - Consistent validation and serialization behavior - Better IDE support and type checking - Clearer separation of concerns by layer - Reduced maintenance cost from mixed validation approaches Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
140 lines
4.0 KiB
Python
140 lines
4.0 KiB
Python
"""Geo and IP lookup Pydantic models.
|
|
|
|
Response models for the ``GET /api/geo/lookup/{ip}`` endpoint.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Awaitable, Callable
|
|
from dataclasses import dataclass
|
|
from typing import TYPE_CHECKING
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
if TYPE_CHECKING:
|
|
import aiohttp
|
|
import aiosqlite
|
|
|
|
|
|
class GeoDetail(BaseModel):
|
|
"""Enriched geolocation data for an IP address.
|
|
|
|
Populated from the ip-api.com free API.
|
|
"""
|
|
|
|
model_config = ConfigDict(strict=True)
|
|
|
|
country_code: str | None = Field(
|
|
default=None,
|
|
description="ISO 3166-1 alpha-2 country code.",
|
|
)
|
|
country_name: str | None = Field(
|
|
default=None,
|
|
description="Human-readable country name.",
|
|
)
|
|
asn: str | None = Field(
|
|
default=None,
|
|
description="Autonomous System Number (e.g. ``'AS3320'``).",
|
|
)
|
|
org: str | None = Field(
|
|
default=None,
|
|
description="Organisation associated with the ASN.",
|
|
)
|
|
|
|
|
|
class GeoCacheEntry(BaseModel):
|
|
"""A single cached geolocation entry for an IP address.
|
|
|
|
Represents a row from the ``geo_cache`` table in the application database.
|
|
"""
|
|
|
|
model_config = ConfigDict(strict=True)
|
|
|
|
ip: str = Field(..., description="IP address (IPv4 or IPv6).")
|
|
country_code: str | None = Field(
|
|
default=None,
|
|
description="ISO 3166-1 alpha-2 country code.",
|
|
)
|
|
country_name: str | None = Field(
|
|
default=None,
|
|
description="Human-readable country name.",
|
|
)
|
|
asn: str | None = Field(
|
|
default=None,
|
|
description="Autonomous System Number (e.g. ``'AS3320'``).",
|
|
)
|
|
org: str | None = Field(
|
|
default=None,
|
|
description="Organisation associated with the ASN.",
|
|
)
|
|
|
|
|
|
class GeoCacheStatsResponse(BaseModel):
|
|
"""Response for ``GET /api/geo/stats``.
|
|
|
|
Exposes diagnostic counters of the geo cache subsystem so operators
|
|
can assess resolution health from the UI or CLI.
|
|
"""
|
|
|
|
model_config = ConfigDict(strict=True)
|
|
|
|
cache_size: int = Field(..., description="Number of positive entries in the in-memory cache.")
|
|
unresolved: int = Field(..., description="Number of geo_cache rows with country_code IS NULL.")
|
|
neg_cache_size: int = Field(..., description="Number of entries in the in-memory negative cache.")
|
|
dirty_size: int = Field(..., description="Number of newly resolved entries not yet flushed to disk.")
|
|
|
|
|
|
class GeoReResolveResponse(BaseModel):
|
|
"""Response for ``POST /api/geo/re-resolve``.
|
|
|
|
Reports how many previously unresolved IPs were retried and how many
|
|
gained a resolved country code after the re-resolve operation.
|
|
"""
|
|
|
|
model_config = ConfigDict(strict=True)
|
|
|
|
resolved: int = Field(..., description="Number of IPs successfully resolved.")
|
|
total: int = Field(..., description="Number of IPs retried.")
|
|
|
|
|
|
class IpLookupResponse(BaseModel):
|
|
"""Response for ``GET /api/geo/lookup/{ip}``.
|
|
|
|
Aggregates current ban status and geographical information for an IP.
|
|
"""
|
|
|
|
model_config = ConfigDict(strict=True)
|
|
|
|
ip: str = Field(..., description="The queried IP address.")
|
|
currently_banned_in: list[str] = Field(
|
|
default_factory=list,
|
|
description="Names of jails where this IP is currently banned.",
|
|
)
|
|
geo: GeoDetail | None = Field(
|
|
default=None,
|
|
description="Enriched geographical and network information.",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# shared service types
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class GeoInfo:
|
|
"""Geo resolution result used throughout backend services."""
|
|
|
|
country_code: str | None
|
|
country_name: str | None
|
|
asn: str | None
|
|
org: str | None
|
|
|
|
|
|
GeoEnricher = Callable[[str], Awaitable[GeoInfo | None]]
|
|
GeoBatchLookup = Callable[
|
|
[list[str], "aiohttp.ClientSession", "aiosqlite.Connection | None"],
|
|
Awaitable[dict[str, GeoInfo]],
|
|
]
|
|
GeoCacheLookup = Callable[[list[str]], tuple[dict[str, GeoInfo], list[str]]]
|