Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .github/workflows/test.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,9 @@ jobs:
- name: Install dependencies
run: |
make install
- name: Run lint
run: |
make lint RUFF_FLAGS=--output-format=github
- name: Run tests
run: |
make test PYREFLY_FLAGS=--output-format=github
4 changes: 4 additions & 0 deletions .mise.toml
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,10 @@ run = "uv venv .venv --allow-existing --seed"
depends = ["venv"]
run = "uv pip install -e '.[dev]' --python .venv/bin/python"

[tasks.lint]
depends = ["install"]
run = ".venv/bin/ruff check ."

[tasks.test]
depends = ["install"]
run = ".venv/bin/pyrefly check --summarize-errors && .venv/bin/pytest -vv"
Expand Down
6 changes: 6 additions & 0 deletions Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,9 @@ UVICORN=$(VENV)/bin/uvicorn
RUFF=$(VENV)/bin/ruff
PACKAGE=dddpy
PYREFLY_FLAGS=--summarize-errors
RUFF_FLAGS=

.PHONY: sync venv install lint typecheck test format dev

sync:
uv sync
Expand All @@ -15,6 +18,9 @@ venv:
install: venv
uv pip install -e ".[dev]" --python $(VENV)/bin/python

lint: install
$(RUFF) check . $(RUFF_FLAGS)

typecheck: install
$(PYREFLY) check $(PYREFLY_FLAGS)

Expand Down
124 changes: 37 additions & 87 deletions dddpy/domain/todo/entities/todo.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Define the Todo entity used throughout the domain layer."""

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

from dddpy.domain.todo.value_objects import (
TodoDescription,
Expand All @@ -10,133 +10,85 @@
TodoTitle,
)

ALREADY_COMPLETED_ERROR_MESSAGE = 'Already completed'


@dataclass(eq=False)
class Todo:
"""Represent a todo item tracked by the domain.

Attributes:
_id: Unique identifier for the todo.
_title: Title describing the todo.
_description: Optional detailed description.
_status: Current lifecycle status.
_created_at: Timestamp when the todo was created.
_updated_at: Timestamp when the todo was last updated.
_completed_at: Optional timestamp when the todo was completed.
id: Unique identifier for the todo.
title: Title describing the todo.
description: Optional detailed description.
status: Current lifecycle status.
created_at: Timestamp when the todo was created.
updated_at: Timestamp when the todo was last updated.
completed_at: Optional timestamp when the todo was completed.
"""

def __init__(
self,
id: TodoId,
title: TodoTitle,
description: Optional[TodoDescription] = None,
status: TodoStatus = TodoStatus.NOT_STARTED,
created_at: datetime = datetime.now(),
updated_at: datetime = datetime.now(),
completed_at: Optional[datetime] = None,
):
"""Initialize a todo domain entity.
id: TodoId
title: TodoTitle
description: TodoDescription | None = None
status: TodoStatus = TodoStatus.NOT_STARTED
created_at: datetime = field(default_factory=datetime.now)
updated_at: datetime = field(default_factory=datetime.now)
completed_at: datetime | None = None

Args:
id: Identifier for the todo.
title: Title describing the todo.
description: Optional longer description.
status: Initial lifecycle status.
created_at: Creation timestamp in UTC.
updated_at: Last updated timestamp in UTC.
completed_at: Optional completion timestamp in UTC.
"""
self._id = id
self._title = title
self._description = description
self._status = status
self._created_at = created_at
self._updated_at = updated_at
self._completed_at = completed_at
def __hash__(self) -> int:
"""Return a hash value based on the entity identity."""
return hash(self.id)

def __eq__(self, obj: object) -> bool:
"""Compare todos by identifier."""
if isinstance(obj, Todo):
return self.id == obj.id

return False

@property
def id(self) -> TodoId:
"""Return the todo's unique identifier."""
return self._id

@property
def title(self) -> TodoTitle:
"""Return the todo's title."""
return self._title

@property
def description(self) -> Optional[TodoDescription]:
"""Return the todo's description if available."""
return self._description

@property
def status(self) -> TodoStatus:
"""Return the todo's current status."""
return self._status

@property
def created_at(self) -> datetime:
"""Return the todo's creation timestamp."""
return self._created_at

@property
def updated_at(self) -> datetime:
"""Return the todo's last update timestamp."""
return self._updated_at

@property
def completed_at(self) -> Optional[datetime]:
"""Return the todo's completion timestamp if set."""
return self._completed_at

def update_title(self, new_title: TodoTitle) -> None:
"""Update the todo title and refresh timestamps.

Args:
new_title: Replacement title for the todo.
"""
self._title = new_title
self._updated_at = datetime.now()
self.title = new_title
self.updated_at = datetime.now()

def update_description(self, new_description: Optional[TodoDescription]) -> None:
def update_description(self, new_description: TodoDescription | None) -> None:
"""Update the todo description and refresh timestamps.

Args:
new_description: Optional replacement description.
"""
self._description = new_description if new_description else None
self._updated_at = datetime.now()
self.description = new_description if new_description else None
self.updated_at = datetime.now()

def start(self) -> None:
"""Mark the todo as in progress and update timestamps."""
self._status = TodoStatus.IN_PROGRESS
self._updated_at = datetime.now()
self.status = TodoStatus.IN_PROGRESS
self.updated_at = datetime.now()

def complete(self) -> None:
"""Mark the todo as completed and record completion time.

Raises:
ValueError: If the todo is already completed.
"""
if self._status == TodoStatus.COMPLETED:
raise ValueError('Already completed')
if self.status == TodoStatus.COMPLETED:
raise ValueError(ALREADY_COMPLETED_ERROR_MESSAGE)

self._status = TodoStatus.COMPLETED
self._completed_at = datetime.now()
self._updated_at = self._completed_at
self.status = TodoStatus.COMPLETED
self.completed_at = datetime.now()
self.updated_at = self.completed_at

@property
def is_completed(self) -> bool:
"""Return whether the todo is marked as completed."""
return self._status == TodoStatus.COMPLETED
return self.status == TodoStatus.COMPLETED

def is_overdue(
self, deadline: datetime, current_time: Optional[datetime] = None
self, deadline: datetime, current_time: datetime | None = None
) -> bool:
"""Determine whether the todo has passed the provided deadline.

Expand All @@ -152,9 +104,7 @@ def is_overdue(
return (current_time or datetime.now()) > deadline

@staticmethod
def create(
title: TodoTitle, description: Optional[TodoDescription] = None
) -> 'Todo':
def create(title: TodoTitle, description: TodoDescription | None = None) -> 'Todo':
"""Create a new todo entity with generated identifier.

Args:
Expand Down
5 changes: 2 additions & 3 deletions dddpy/domain/todo/repositories/todo_repository.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
"""Define the repository abstraction for todo entities."""

from abc import ABC, abstractmethod
from typing import List, Optional

from dddpy.domain.todo.entities import Todo
from dddpy.domain.todo.value_objects import TodoId
Expand All @@ -19,7 +18,7 @@ def save(self, todo: Todo) -> None:
"""

@abstractmethod
def find_by_id(self, todo_id: TodoId) -> Optional[Todo]:
def find_by_id(self, todo_id: TodoId) -> Todo | None:
"""Retrieve a todo by its identifier.

Args:
Expand All @@ -30,7 +29,7 @@ def find_by_id(self, todo_id: TodoId) -> Optional[Todo]:
"""

@abstractmethod
def find_all(self) -> List[Todo]:
def find_all(self) -> list[Todo]:
"""Return the collection of todos stored in the repository.

Returns:
Expand Down
7 changes: 5 additions & 2 deletions dddpy/domain/todo/value_objects/todo_description.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,9 @@

from dataclasses import dataclass

MAX_DESCRIPTION_LENGTH = 1000
DESCRIPTION_TOO_LONG_ERROR_MESSAGE = 'Description must be 1000 characters or less'


@dataclass(frozen=True)
class TodoDescription:
Expand All @@ -15,8 +18,8 @@ def __post_init__(self):
Raises:
ValueError: If the description exceeds 1000 characters.
"""
if len(self.value) > 1000:
raise ValueError('Description must be 1000 characters or less')
if len(self.value) > MAX_DESCRIPTION_LENGTH:
raise ValueError(DESCRIPTION_TOO_LONG_ERROR_MESSAGE)

def __str__(self) -> str:
"""Return the wrapped description string."""
Expand Down
10 changes: 7 additions & 3 deletions dddpy/domain/todo/value_objects/todo_title.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,10 @@

from dataclasses import dataclass

MAX_TITLE_LENGTH = 100
TITLE_REQUIRED_ERROR_MESSAGE = 'Title is required'
TITLE_TOO_LONG_ERROR_MESSAGE = 'Title must be 100 characters or less'


@dataclass(frozen=True)
class TodoTitle:
Expand All @@ -16,9 +20,9 @@ def __post_init__(self):
ValueError: If the title is empty or longer than 100 characters.
"""
if not self.value:
raise ValueError('Title is required')
if len(self.value) > 100:
raise ValueError('Title must be 100 characters or less')
raise ValueError(TITLE_REQUIRED_ERROR_MESSAGE)
if len(self.value) > MAX_TITLE_LENGTH:
raise ValueError(TITLE_TOO_LONG_ERROR_MESSAGE)

def __str__(self) -> str:
"""Return the wrapped title string."""
Expand Down
2 changes: 1 addition & 1 deletion dddpy/infrastructure/di/injection.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Dependency injection configuration for the application."""

from typing import Iterator
from collections.abc import Iterator

from fastapi import Depends
from sqlalchemy.orm import Session
Expand Down
6 changes: 2 additions & 4 deletions dddpy/infrastructure/sqlite/todo/todo_repository.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
"""SQLite implementation of Todo repository."""

from typing import List, Optional

from sqlalchemy import desc
from sqlalchemy.exc import NoResultFound
from sqlalchemy.orm.session import Session
Expand All @@ -23,7 +21,7 @@ def __init__(self, session: Session):
"""
self.session = session

def find_by_id(self, todo_id: TodoId) -> Optional[Todo]:
def find_by_id(self, todo_id: TodoId) -> Todo | None:
"""Return a todo matching the provided identifier.

Args:
Expand All @@ -39,7 +37,7 @@ def find_by_id(self, todo_id: TodoId) -> Optional[Todo]:

return row.to_entity()

def find_all(self) -> List[Todo]:
def find_all(self) -> list[Todo]:
"""Return todos ordered by creation date with an upper limit.

Returns:
Expand Down
Loading
Loading