saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
30 lines
1.2 KiB
Python
30 lines
1.2 KiB
Python
"""Per-IP, per-action escalation state for the throttle/ban engine
|
|
(app/services/ip_throttle_service.py). `offense_count` is the "memory" that
|
|
survives a completed timeout cycle - only a manual admin unban resets it."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import DateTime, Enum, Integer, String, UniqueConstraint
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from app.db.base import Base, TimestampMixin, UUIDPrimaryKeyMixin
|
|
from app.models.enums import ThrottleAction
|
|
|
|
|
|
class IpThrottleState(Base, UUIDPrimaryKeyMixin, TimestampMixin):
|
|
__tablename__ = "ip_throttle_state"
|
|
__table_args__ = (
|
|
UniqueConstraint("ip_address", "action", name="uq_ip_throttle_state_ip_action"),
|
|
)
|
|
|
|
ip_address: Mapped[str] = mapped_column(String(45), index=True)
|
|
action: Mapped[ThrottleAction] = mapped_column(
|
|
Enum(ThrottleAction, native_enum=False, length=24)
|
|
)
|
|
attempt_count: Mapped[int] = mapped_column(Integer, default=0)
|
|
next_allowed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
|
|
timeout_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
|
|
offense_count: Mapped[int] = mapped_column(Integer, default=0)
|