Recommended resources
The freely available resources this lesson will direct you to.
Hello! In the previous lesson, you assigned responsibilities to the note reader, review services, scheduling policy, repository, and Telegram adapter. The next question is what those components pass between one another. If a “note” means a database row in one component and an API response in another, the boundaries will be difficult to maintain.
This lesson defines four Python domain models: a note, an immutable review event, scheduling state, and a reminder attempt. They express the system’s concepts without importing FastAPI, PostgreSQL drivers, or Telegram libraries. Allow about 40 minutes: roughly three minutes of video, five minutes of reading, and the remainder to work through the design and code.
Model the concepts before their storage
A domain model describes something the application needs to reason about. A database schema describes how to store it; an API schema describes how to exchange it over HTTP. Those descriptions may contain similar fields, but they have different responsibilities.
Consider a note renamed in the synchronized vault. Its file path changes, but its identity should not. A domain Note therefore needs a stable note_id as well as a source_path. Neither a PostgreSQL primary-key declaration nor a FastAPI request model should determine what that distinction means.
The repository-pattern diagram below captures the boundary we want. On its “after” side, database access is outside the domain. It illustrates a dependency principle, not a requirement to use the SQLAlchemy implementation pictured: this course will later use a PostgreSQL repository with psycopg.
For a short explanation, read the “Applying the DIP to Data Access” subsection of Repository Pattern. Focus on why the domain model should not need infrastructure to run or be tested.
Read the “Applying the DIP to Data Access” subsection to understand the intended direction of dependency.
In “Applying the DIP to Data Access,” start with domain independence. Continue through the following paragraph beginning “Instead, as discussed in the introduction,” and stop before the “Figure 3. Onion architecture” diagram. Notice that the presentation and database layers can know about domain concepts without making the domain import either layer.
This boundary does not mean domain objects can contain no rules. “A review count cannot be negative” is a domain invariant. “This JSON field is required” is an API concern; “this column is indexed” is a persistence concern. Keeping those distinctions clear lets the same domain objects participate in HTTP and Telegram workflows.
Use dataclasses, but know what they provide
Python’s standard-library @dataclass generates an initializer, readable representation, and value-based equality from annotated fields. Setting frozen=True prevents ordinary reassignment of those fields. Neither feature, by itself, validates that an identifier is meaningful or a timestamp has a timezone.
Watch mCoding’s Python dataclasses will save you HOURS, also featuring attrs for the generated methods, frozen instances, and the mutable-default pitfall. We will use a tuple for note tags so an instance cannot acquire a shared, mutable default list.
Python dataclasses will save you HOURS, also featuring attrs
Watch mCoding’s dataclass walkthrough for the mechanics behind the models below.
Watch the dataclass walkthrough. Focus on the generated initializer and representation, what frozen=True prevents, and why a mutable collection requires care when used as a default. Ordering and hashing are not requirements for our four models.
The Python documentation explains one additional tool we will use: __post_init__ runs after the generated initializer. It is a suitable place for small, framework-independent invariants.
Use the Python standard-library documentation to check what a dataclass generates and when post-initialization checks run.
In the opening of “dataclasses — Data Classes,” read the first example, ending with the explanation after the generated initializer. Then go to “Post-init processing” and read post-init behavior. Concentrate on the timing of __post_init__, rather than the inheritance example that follows.
Give each model one meaning
Before writing the classes, settle what each object represents.
A Note is the current ingested content of one source note. Its note_id remains stable across file renames; source_path records where the current file came from. title, body, and tags are content, not scheduling data. In this proposal, an empty body is allowed, while an empty title is not. An ingestion policy can make a stricter decision later without mistaking that policy for a dataclass feature.
A ReviewEvent is one accepted recall submission. Its event_id identifies the recorded event; its submission_key identifies the logical submission so a retry can be recognized. It retains the note ID, rating, and time of review. An event is a historical fact: accepting a later review creates another event rather than editing the earlier one. This proposal supports AGAIN, HARD, GOOD, and EASY ratings; if the workflow contract chooses different ratings, change this enum before implementing the scheduling policy.
SchedulingState is the latest state used to determine a note’s next review. It contains a due time, last-review time, review count, and current interval. For a newly ingested note, the count and interval are zero and there is no last-review time. The model does not yet decide what interval a rating earns—that is the scheduling-policy lesson later in the course.
Finally, a ReminderAttempt represents the durable claim for one recipient and reminder window. No record means that window has not been claimed; an existing record means another run must not initiate a second request under our at-most-one-request design. REQUEST_ACCEPTED means the Telegram API accepted a request, not that the person received or read a message. REQUEST_OUTCOME_UNKNOWN covers an ambiguous result such as a timeout. Neither status authorizes a retry for that window.
These four objects are not four database tables by decree. The schema lesson will decide how to persist them and which database constraints enforce cross-record guarantees.
A framework-independent Python definition
The following can live in a domain module with only standard-library imports. It uses Python 3.10 or later for kw_only=True; keyword-only construction makes call sites easier to read when several fields are strings or timestamps.
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
def require_aware(value: datetime, field_name: str) -> None:
"""Reject naive timestamps, whose timezone is unspecified."""
if not isinstance(value, datetime) or value.utcoffset() is None:
raise ValueError(f"{field_name} must be a timezone-aware datetime")
class RecallRating(str, Enum):
AGAIN = "again"
HARD = "hard"
GOOD = "good"
EASY = "easy"
class ReminderStatus(str, Enum):
CLAIMED = "claimed"
REQUEST_ACCEPTED = "request_accepted"
REQUEST_OUTCOME_UNKNOWN = "request_outcome_unknown"
@dataclass(frozen=True, kw_only=True)
class Note:
note_id: str
source_path: str
title: str
body: str
tags: tuple[str, ...] = ()
def __post_init__(self) -> None:
if not self.note_id.strip():
raise ValueError("note_id must not be blank")
if not self.source_path.strip():
raise ValueError("source_path must not be blank")
if not self.title.strip():
raise ValueError("title must not be blank")
if not isinstance(self.tags, tuple) or any(
not isinstance(tag, str) for tag in self.tags
):
raise ValueError("tags must be a tuple of strings")
@dataclass(frozen=True, kw_only=True)
class ReviewEvent:
event_id: str
submission_key: str
note_id: str
rating: RecallRating
reviewed_at: datetime
def __post_init__(self) -> None:
if not self.event_id.strip():
raise ValueError("event_id must not be blank")
if not self.submission_key.strip():
raise ValueError("submission_key must not be blank")
if not self.note_id.strip():
raise ValueError("note_id must not be blank")
if not isinstance(self.rating, RecallRating):
raise ValueError("rating must be a RecallRating")
require_aware(self.reviewed_at, "reviewed_at")
@dataclass(frozen=True, kw_only=True)
class SchedulingState:
note_id: str
due_at: datetime
last_reviewed_at: datetime | None = None
review_count: int = 0
interval_days: int = 0
def __post_init__(self) -> None:
if not self.note_id.strip():
raise ValueError("note_id must not be blank")
require_aware(self.due_at, "due_at")
if self.last_reviewed_at is not None:
require_aware(self.last_reviewed_at, "last_reviewed_at")
if type(self.review_count) is not int or self.review_count < 0:
raise ValueError("review_count must be a nonnegative integer")
if type(self.interval_days) is not int or self.interval_days < 0:
raise ValueError("interval_days must be a nonnegative integer")
if (self.review_count == 0) != (self.last_reviewed_at is None):
raise ValueError(
"last_reviewed_at must be absent exactly when "
"review_count is zero"
)
@dataclass(frozen=True, kw_only=True)
class ReminderAttempt:
recipient_id: str
window_start: datetime
window_end: datetime
claimed_at: datetime
status: ReminderStatus
request_accepted_at: datetime | None = None
def __post_init__(self) -> None:
if not self.recipient_id.strip():
raise ValueError("recipient_id must not be blank")
for field_name in ("window_start", "window_end", "claimed_at"):
require_aware(getattr(self, field_name), field_name)
if self.window_end <= self.window_start:
raise ValueError("window_end must follow window_start")
if not isinstance(self.status, ReminderStatus):
raise ValueError("status must be a ReminderStatus")
if self.request_accepted_at is not None:
require_aware(self.request_accepted_at, "request_accepted_at")
if (self.status is ReminderStatus.REQUEST_ACCEPTED) != (
self.request_accepted_at is not None
):
raise ValueError(
"request_accepted_at is required only for REQUEST_ACCEPTED"
)
Notice the deliberately narrow checks. __post_init__ rejects internally contradictory objects, such as a first-review state that claims to have a last-review timestamp. It does not parse Markdown, interpret a Telegram callback, or choose a scheduling interval. Type annotations also do not perform automatic runtime conversion: the input-facing code must convert incoming values to the appropriate strings, enum members, and datetime objects.
frozen=True is particularly helpful for historical ReviewEvent instances. For SchedulingState, it encourages the future scheduling function to return a new state rather than quietly mutating a shared object. Freezing does not make every value deeply immutable; the tuple of tag strings avoids the most immediate collection problem.
Read the models against a real workflow
Suppose a synchronized file produces note ID n-42. A note object can carry its current path and content while a separate state says when that note is due:
from datetime import datetime, timezone
note = Note(
note_id="n-42",
source_path="topics/algebra.md",
title="Algebra",
body="Recall the key definitions before checking the note.",
tags=("mathematics",),
)
state = SchedulingState(
note_id=note.note_id,
due_at=datetime(2026, 3, 10, 9, tzinfo=timezone.utc),
)
event = ReviewEvent(
event_id="event-001",
submission_key="submission-001",
note_id=note.note_id,
rating=RecallRating.GOOD,
reviewed_at=datetime(2026, 3, 10, 9, 5, tzinfo=timezone.utc),
)
The event does not itself update state. An application service will later check whether the submission key has already been accepted, ask the pure scheduling policy for a new state, and persist the event and new state together. Keeping them separate preserves both the historical answer—what happened?—and the current answer—when is this note due now?
Likewise, constructing a reminder claim is not equivalent to sending a message:
claim = ReminderAttempt(
recipient_id="configured-recipient",
window_start=datetime(2026, 3, 10, 9, tzinfo=timezone.utc),
window_end=datetime(2026, 3, 11, 9, tzinfo=timezone.utc),
claimed_at=datetime(2026, 3, 10, 9, tzinfo=timezone.utc),
status=ReminderStatus.CLAIMED,
)
The intended window identity is the pair recipient_id and window_start. This class can validate one claim, but it cannot detect a second claim stored elsewhere. The eventual repository and PostgreSQL constraint must enforce uniqueness, and the service must claim the window before initiating the external request. If the process crashes after the claim but before sending, the at-most-one-request design may produce no reminder for that window. That is the trade-off, not a defect that the domain object can conceal.
The same distinction applies to review retries: ReviewEvent carries a submission_key, but an object in memory cannot prove that no other event has that key. Domain invariants protect each object; services and persistence enforce rules spanning multiple objects and operations.
Conclusion
You now have four domain concepts with distinct roles: Note represents current source content, ReviewEvent preserves accepted feedback, SchedulingState represents the latest review position, and ReminderAttempt records a claim and its request status. Their definitions use standard Python, meaningful identifiers, aware timestamps, and small invariants—without depending on an API framework or database.
Next, the PostgreSQL schema lesson will translate these concepts into stored records, keys, constraints, and indexes. That translation should preserve the meanings established here rather than letting table structure redefine them.