chore: add changelog and documentation updates
All checks were successful
CI / lint-and-test (push) Successful in 17s

- Created a new `CHANGELOG.md` file to document all notable changes to the project, adhering to the Keep a Changelog format.
- Updated `CONTRIBUTING.md` to include instructions for building and previewing documentation using MkDocs.
- Added `mkdocs.yml` configuration for documentation generation, including navigation structure and theme settings.
- Enhanced various documentation files, including API reference, architecture overview, configuration reference, and runbook, to provide comprehensive guidance for users and developers.
- Included new sections in the README for changelog and documentation links, improving accessibility to project information.
This commit is contained in:
2026-02-20 15:32:10 +03:00
parent b61e1ca8a5
commit 86f6d66865
88 changed files with 28912 additions and 118 deletions

View File

@@ -49,6 +49,12 @@ __all__ = [
def init_db(database_url: str) -> None:
"""Create tables from metadata (Alembic migrations handle schema in production)."""
"""Create all tables from SQLAlchemy metadata.
Prefer Alembic migrations for schema changes in production.
Args:
database_url: SQLAlchemy database URL.
"""
engine = get_engine(database_url)
Base.metadata.create_all(bind=engine)

View File

@@ -11,6 +11,8 @@ class Base(DeclarativeBase):
class User(Base):
"""Telegram user and display name; may have telegram_user_id=None for import-only users."""
__tablename__ = "users"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
@@ -43,6 +45,8 @@ class CalendarSubscriptionToken(Base):
class Duty(Base):
"""Single duty/unavailable/vacation slot (UTC start_at/end_at, event_type)."""
__tablename__ = "duties"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

View File

@@ -11,7 +11,15 @@ from duty_teller.db.models import User, Duty, GroupDutyPin, CalendarSubscription
def get_user_by_telegram_id(session: Session, telegram_user_id: int) -> User | None:
"""Find user by telegram_user_id. Returns None if not found (no creation)."""
"""Find user by Telegram user ID.
Args:
session: DB session.
telegram_user_id: Telegram user id.
Returns:
User or None if not found. Does not create a user.
"""
return session.query(User).filter(User.telegram_user_id == telegram_user_id).first()
@@ -23,10 +31,22 @@ def get_or_create_user(
first_name: str | None = None,
last_name: str | None = None,
) -> User:
"""
Get or create user by telegram_user_id. On create, name comes from Telegram.
On update: username is always synced; full_name/first_name/last_name are only
updated if name_manually_edited is False (otherwise keep existing display name).
"""Get or create user by Telegram user ID.
On create, name fields come from Telegram. On update: username is always
synced; full_name, first_name, last_name are updated only if
name_manually_edited is False (otherwise existing display name is kept).
Args:
session: DB session.
telegram_user_id: Telegram user id.
full_name: Display full name.
username: Telegram username (optional).
first_name: Telegram first name (optional).
last_name: Telegram last name (optional).
Returns:
User instance (created or updated).
"""
user = get_user_by_telegram_id(session, telegram_user_id)
if user:
@@ -53,9 +73,16 @@ def get_or_create_user(
def get_or_create_user_by_full_name(session: Session, full_name: str) -> User:
"""
Find user by exact full_name or create one with telegram_user_id=None (for duty-schedule import).
New users get name_manually_edited=True since the name comes from import, not Telegram.
"""Find user by exact full_name or create one (for duty-schedule import).
New users have telegram_user_id=None and name_manually_edited=True.
Args:
session: DB session.
full_name: Exact full name to match or set.
Returns:
User instance (existing or newly created).
"""
user = session.query(User).filter(User.full_name == full_name).first()
if user:
@@ -81,10 +108,20 @@ def update_user_display_name(
first_name: str | None = None,
last_name: str | None = None,
) -> User | None:
"""
Update display name for user by telegram_user_id and set name_manually_edited=True.
Use from API or admin when name is changed manually; subsequent get_or_create_user
will not overwrite these fields. Returns User or None if not found.
"""Update display name and set name_manually_edited=True.
Use from API or admin when name is changed manually; subsequent
get_or_create_user will not overwrite these fields.
Args:
session: DB session.
telegram_user_id: Telegram user id.
full_name: New full name.
first_name: New first name (optional).
last_name: New last name (optional).
Returns:
Updated User or None if not found.
"""
user = session.query(User).filter(User.telegram_user_id == telegram_user_id).first()
if not user:
@@ -104,7 +141,17 @@ def delete_duties_in_range(
from_date: str,
to_date: str,
) -> int:
"""Delete all duties of the user that overlap [from_date, to_date] (YYYY-MM-DD). Returns count deleted."""
"""Delete all duties of the user that overlap the given date range.
Args:
session: DB session.
user_id: User id.
from_date: Start date YYYY-MM-DD.
to_date: End date YYYY-MM-DD.
Returns:
Number of duties deleted.
"""
to_next = (
datetime.fromisoformat(to_date + "T00:00:00") + timedelta(days=1)
).strftime("%Y-%m-%d")
@@ -124,7 +171,16 @@ def get_duties(
from_date: str,
to_date: str,
) -> list[tuple[Duty, str]]:
"""Return list of (Duty, full_name) overlapping the given date range."""
"""Return duties overlapping the given date range with user full_name.
Args:
session: DB session.
from_date: Start date YYYY-MM-DD.
to_date: End date YYYY-MM-DD.
Returns:
List of (Duty, full_name) tuples.
"""
to_date_next = (
datetime.fromisoformat(to_date + "T00:00:00") + timedelta(days=1)
).strftime("%Y-%m-%d")
@@ -142,7 +198,17 @@ def get_duties_for_user(
from_date: str,
to_date: str,
) -> list[tuple[Duty, str]]:
"""Return list of (Duty, full_name) for the given user overlapping the date range."""
"""Return duties for one user overlapping the date range.
Args:
session: DB session.
user_id: User id.
from_date: Start date YYYY-MM-DD.
to_date: End date YYYY-MM-DD.
Returns:
List of (Duty, full_name) tuples.
"""
to_date_next = (
datetime.fromisoformat(to_date + "T00:00:00") + timedelta(days=1)
).strftime("%Y-%m-%d")
@@ -164,9 +230,17 @@ def _token_hash(token: str) -> str:
def create_calendar_token(session: Session, user_id: int) -> str:
"""
Create a new calendar subscription token for the user.
Removes any existing tokens for this user. Returns the raw token string.
"""Create a new calendar subscription token for the user.
Any existing tokens for this user are removed. The raw token is returned
only once (not stored in plain text).
Args:
session: DB session.
user_id: User id.
Returns:
Raw token string (e.g. for URL /api/calendar/ical/{token}.ics).
"""
session.query(CalendarSubscriptionToken).filter(
CalendarSubscriptionToken.user_id == user_id
@@ -185,9 +259,16 @@ def create_calendar_token(session: Session, user_id: int) -> str:
def get_user_by_calendar_token(session: Session, token: str) -> User | None:
"""
Find user by calendar subscription token. Uses constant-time comparison.
Returns None if token is invalid or not found.
"""Find user by calendar subscription token.
Uses constant-time comparison to avoid timing leaks.
Args:
session: DB session.
token: Raw token from URL.
Returns:
User or None if token is invalid or not found.
"""
token_hash_val = _token_hash(token)
row = (
@@ -211,7 +292,18 @@ def insert_duty(
end_at: str,
event_type: str = "duty",
) -> Duty:
"""Create a duty. start_at and end_at must be UTC, ISO 8601 with Z."""
"""Create a duty record.
Args:
session: DB session.
user_id: User id.
start_at: Start time UTC, ISO 8601 with Z (e.g. 2025-01-15T09:00:00Z).
end_at: End time UTC, ISO 8601 with Z.
event_type: One of "duty", "unavailable", "vacation". Default "duty".
Returns:
Created Duty instance.
"""
duty = Duty(
user_id=user_id,
start_at=start_at,
@@ -225,7 +317,15 @@ def insert_duty(
def get_current_duty(session: Session, at_utc: datetime) -> tuple[Duty, User] | None:
"""Return the duty (and user) for which start_at <= at_utc < end_at, event_type='duty'."""
"""Return the duty and user active at the given UTC time (event_type='duty').
Args:
session: DB session.
at_utc: Point in time (timezone-aware or naive UTC).
Returns:
(Duty, User) or None if no duty at that time.
"""
from datetime import timezone
if at_utc.tzinfo is not None:
@@ -247,7 +347,15 @@ def get_current_duty(session: Session, at_utc: datetime) -> tuple[Duty, User] |
def get_next_shift_end(session: Session, after_utc: datetime) -> datetime | None:
"""Return the end_at of the current duty or of the next duty. Naive UTC."""
"""Return the end_at of the current or next duty (event_type='duty').
Args:
session: DB session.
after_utc: Point in time (timezone-aware or naive UTC).
Returns:
End datetime (naive UTC) or None if no current or future duty.
"""
from datetime import timezone
if after_utc.tzinfo is not None:
@@ -280,14 +388,31 @@ def get_next_shift_end(session: Session, after_utc: datetime) -> datetime | None
def get_group_duty_pin(session: Session, chat_id: int) -> GroupDutyPin | None:
"""Get the pinned message record for a chat, if any."""
"""Get the pinned duty message record for a chat.
Args:
session: DB session.
chat_id: Telegram chat id.
Returns:
GroupDutyPin or None.
"""
return session.query(GroupDutyPin).filter(GroupDutyPin.chat_id == chat_id).first()
def save_group_duty_pin(
session: Session, chat_id: int, message_id: int
) -> GroupDutyPin:
"""Save or update the pinned message for a chat."""
"""Save or update the pinned duty message for a chat.
Args:
session: DB session.
chat_id: Telegram chat id.
message_id: Message id to pin/update.
Returns:
GroupDutyPin instance (created or updated).
"""
pin = session.query(GroupDutyPin).filter(GroupDutyPin.chat_id == chat_id).first()
if pin:
pin.message_id = message_id
@@ -300,13 +425,27 @@ def save_group_duty_pin(
def delete_group_duty_pin(session: Session, chat_id: int) -> None:
"""Remove the pinned message record when the bot leaves the group."""
"""Remove the pinned duty message record for the chat (e.g. when bot leaves group).
Args:
session: DB session.
chat_id: Telegram chat id.
"""
session.query(GroupDutyPin).filter(GroupDutyPin.chat_id == chat_id).delete()
session.commit()
def get_all_group_duty_pin_chat_ids(session: Session) -> list[int]:
"""Return all chat_ids that have a pinned duty message (for restoring jobs on startup)."""
"""Return all chat_ids that have a pinned duty message.
Used to restore update jobs on bot startup.
Args:
session: DB session.
Returns:
List of chat ids.
"""
rows = session.query(GroupDutyPin.chat_id).all()
return [r[0] for r in rows]
@@ -314,7 +453,16 @@ def get_all_group_duty_pin_chat_ids(session: Session) -> list[int]:
def set_user_phone(
session: Session, telegram_user_id: int, phone: str | None
) -> User | None:
"""Set phone for user by telegram_user_id. Returns User or None if not found."""
"""Set or clear phone for user by Telegram user id.
Args:
session: DB session.
telegram_user_id: Telegram user id.
phone: Phone string or None to clear.
Returns:
Updated User or None if not found.
"""
user = session.query(User).filter(User.telegram_user_id == telegram_user_id).first()
if not user:
return None

View File

@@ -1,4 +1,4 @@
"""Pydantic schemas for API and validation."""
"""Pydantic schemas for API request/response and validation."""
from typing import Literal
@@ -6,6 +6,8 @@ from pydantic import BaseModel, ConfigDict
class UserBase(BaseModel):
"""Base user fields (full_name, username, first/last name)."""
full_name: str
username: str | None = None
first_name: str | None = None
@@ -13,10 +15,14 @@ class UserBase(BaseModel):
class UserCreate(UserBase):
"""User creation payload including Telegram user id."""
telegram_user_id: int
class UserInDb(UserBase):
"""User as stored in DB (includes id and telegram_user_id)."""
id: int
telegram_user_id: int
@@ -24,16 +30,22 @@ class UserInDb(UserBase):
class DutyBase(BaseModel):
"""Duty fields: user_id, start_at, end_at (UTC ISO 8601 with Z)."""
user_id: int
start_at: str # UTC, ISO 8601 with Z
end_at: str # UTC, ISO 8601 with Z
class DutyCreate(DutyBase):
"""Duty creation payload."""
pass
class DutyInDb(DutyBase):
"""Duty as stored in DB (includes id)."""
id: int
model_config = ConfigDict(from_attributes=True)

View File

@@ -21,7 +21,14 @@ _SessionLocal = None
@contextmanager
def session_scope(database_url: str) -> Generator[Session, None, None]:
"""Context manager: yields a session, rolls back on exception, closes on exit."""
"""Context manager that yields a session; rolls back on exception, closes on exit.
Args:
database_url: SQLAlchemy database URL.
Yields:
Session instance. Caller must not use it after exit.
"""
session = get_session(database_url)
try:
yield session
@@ -33,6 +40,7 @@ def session_scope(database_url: str) -> Generator[Session, None, None]:
def get_engine(database_url: str):
"""Return cached SQLAlchemy engine for the given URL (one per process)."""
global _engine
if _engine is None:
_engine = create_engine(
@@ -46,6 +54,7 @@ def get_engine(database_url: str):
def get_session_factory(database_url: str) -> sessionmaker[Session]:
"""Return cached session factory for the given URL (one per process)."""
global _SessionLocal
if _SessionLocal is None:
engine = get_engine(database_url)
@@ -54,4 +63,5 @@ def get_session_factory(database_url: str) -> sessionmaker[Session]:
def get_session(database_url: str) -> Session:
"""Create a new session from the factory for the given URL."""
return get_session_factory(database_url)()