feat: Build the flashcards model, routes, etc.

This commit is contained in:
wilson 2026-04-09 20:40:11 +01:00
parent 0281caef7c
commit 27f7a7c3f3
9 changed files with 652 additions and 0 deletions

View file

@ -0,0 +1,88 @@
"""add flashcard tables
Revision ID: 0009
Revises: 0008
Create Date: 2026-04-08
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
revision: str = "0009"
down_revision: Union[str, None] = "0008"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"flashcard",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column(
"user_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column(
"bank_entry_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("learnable_word_bank_entry.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("source_lang", sa.Text(), nullable=False),
sa.Column("target_lang", sa.Text(), nullable=False),
sa.Column("prompt_text", sa.Text(), nullable=False),
sa.Column("answer_text", sa.Text(), nullable=False),
sa.Column("prompt_context_text", sa.Text(), nullable=True),
sa.Column("answer_context_text", sa.Text(), nullable=True),
sa.Column("card_direction", sa.Text(), nullable=False),
sa.Column("prompt_modality", sa.Text(), nullable=False, server_default="text"),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
nullable=False,
server_default=sa.func.now(),
),
)
op.create_index("ix_flashcard_user_id", "flashcard", ["user_id"])
op.create_index("ix_flashcard_bank_entry_id", "flashcard", ["bank_entry_id"])
op.create_table(
"flashcard_event",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column(
"flashcard_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("flashcard.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column(
"user_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("event_type", sa.Text(), nullable=False),
sa.Column("user_response", sa.Text(), nullable=True),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
nullable=False,
server_default=sa.func.now(),
),
)
op.create_index("ix_flashcard_event_flashcard_id", "flashcard_event", ["flashcard_id"])
op.create_index("ix_flashcard_event_user_id", "flashcard_event", ["user_id"])
def downgrade() -> None:
op.drop_index("ix_flashcard_event_user_id", table_name="flashcard_event")
op.drop_index("ix_flashcard_event_flashcard_id", table_name="flashcard_event")
op.drop_table("flashcard_event")
op.drop_index("ix_flashcard_bank_entry_id", table_name="flashcard")
op.drop_index("ix_flashcard_user_id", table_name="flashcard")
op.drop_table("flashcard")

View file

@ -0,0 +1,28 @@
from dataclasses import dataclass
from datetime import datetime
@dataclass
class Flashcard:
id: str
user_id: str
bank_entry_id: str
source_lang: str
target_lang: str
prompt_text: str
answer_text: str
prompt_context_text: str | None
answer_context_text: str | None
card_direction: str
prompt_modality: str
created_at: datetime
@dataclass
class FlashcardEvent:
id: str
flashcard_id: str
user_id: str
event_type: str
user_response: str | None
created_at: datetime

View file

@ -0,0 +1,152 @@
import uuid
from ..models.flashcard import Flashcard, FlashcardEvent
from ...outbound.postgres.repositories.flashcard_repository import FlashcardRepository
from ...outbound.postgres.repositories.vocab_repository import VocabRepository
from ...outbound.postgres.repositories.dictionary_repository import DictionaryRepository
VALID_DIRECTIONS = {"target_to_en", "en_to_target"}
VALID_EVENT_TYPES = {"shown", "answered", "skipped"}
class FlashcardService:
"""Generates flashcards from resolved vocab bank entries and records study events.
Flashcard text is derived directly from the dictionary: the lemma headword is the
target-language side and the sense gloss is the English side. Both directions are
created by default.
Usage::
service = FlashcardService(
flashcard_repo=PostgresFlashcardRepository(db),
vocab_repo=PostgresVocabRepository(db),
dict_repo=PostgresDictionaryRepository(db),
)
# Generate both directions for a resolved bank entry
cards = await service.generate_flashcard_from_entry(entry_id)
# Record that the user answered correctly
event = await service.record_flashcard_event(
flashcard_id=cards[0].id,
user_id=user_id,
event_type="answered",
response="banque",
)
"""
def __init__(
self,
flashcard_repo: FlashcardRepository,
vocab_repo: VocabRepository,
dict_repo: DictionaryRepository,
) -> None:
self.flashcard_repo = flashcard_repo
self.vocab_repo = vocab_repo
self.dict_repo = dict_repo
async def generate_flashcard_from_entry(
self,
entry_id: uuid.UUID,
direction: str | None = None,
) -> list[Flashcard]:
"""Create flashcard(s) from a vocab bank entry that has a resolved sense.
Looks up the sense gloss (English meaning) and lemma headword (target-language
word) and creates one card per direction. Pass ``direction`` to generate only
``"target_to_en"`` or ``"en_to_target"``; omit it to create both.
Raises ``ValueError`` if the entry does not exist, has no resolved sense, or
if the underlying sense or lemma rows cannot be found in the dictionary.
Usage::
# Both directions — typical case
cards = await service.generate_flashcard_from_entry(entry_id)
assert len(cards) == 2
# One direction only
cards = await service.generate_flashcard_from_entry(
entry_id, direction="target_to_en"
)
"""
if direction is not None and direction not in VALID_DIRECTIONS:
raise ValueError(f"Invalid direction '{direction}'. Must be one of {VALID_DIRECTIONS}")
entry = await self.vocab_repo.get_entry(entry_id)
if entry is None:
raise ValueError(f"Bank entry {entry_id} not found")
if entry.sense_id is None:
raise ValueError(
"Entry has no resolved sense; disambiguate before generating flashcards"
)
sense = await self.dict_repo.get_sense(uuid.UUID(entry.sense_id))
if sense is None:
raise ValueError(f"Sense {entry.sense_id} not found in dictionary")
lemma = await self.dict_repo.get_lemma(uuid.UUID(sense.lemma_id))
if lemma is None:
raise ValueError(f"Lemma for sense {entry.sense_id} not found in dictionary")
pair = await self.vocab_repo.get_language_pair(uuid.UUID(entry.language_pair_id))
if pair is None:
raise ValueError(f"Language pair {entry.language_pair_id} not found")
user_id = uuid.UUID(entry.user_id)
directions = [direction] if direction else ["target_to_en", "en_to_target"]
flashcards = []
for d in directions:
if d == "target_to_en":
prompt, answer = lemma.headword, sense.gloss
else:
prompt, answer = sense.gloss, lemma.headword
card = await self.flashcard_repo.create_flashcard(
user_id=user_id,
bank_entry_id=entry_id,
source_lang=pair.source_lang,
target_lang=pair.target_lang,
prompt_text=prompt,
answer_text=answer,
card_direction=d,
)
flashcards.append(card)
return flashcards
async def record_flashcard_event(
self,
flashcard_id: uuid.UUID,
user_id: uuid.UUID,
event_type: str,
response: str | None = None,
) -> FlashcardEvent:
"""Record a study event against a flashcard — shown, answered, or skipped.
``response`` is the user's free-text answer and is only meaningful for
``event_type="answered"``; it is stored as-is without grading.
Raises ``ValueError`` for unrecognised event types.
Usage::
event = await service.record_flashcard_event(
flashcard_id=card.id,
user_id=user_id,
event_type="answered",
response="banque",
)
"""
if event_type not in VALID_EVENT_TYPES:
raise ValueError(
f"Invalid event_type '{event_type}'. Must be one of {VALID_EVENT_TYPES}"
)
return await self.flashcard_repo.record_event(
flashcard_id=flashcard_id,
user_id=user_id,
event_type=event_type,
user_response=response,
)

View file

@ -0,0 +1,64 @@
import uuid
from datetime import datetime, timezone
from sqlalchemy import DateTime, ForeignKey, Text
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.dialects.postgresql import UUID
from ..database import Base
class FlashcardEntity(Base):
__tablename__ = "flashcard"
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
bank_entry_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("learnable_word_bank_entry.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
source_lang: Mapped[str] = mapped_column(Text, nullable=False)
target_lang: Mapped[str] = mapped_column(Text, nullable=False)
prompt_text: Mapped[str] = mapped_column(Text, nullable=False)
answer_text: Mapped[str] = mapped_column(Text, nullable=False)
prompt_context_text: Mapped[str | None] = mapped_column(Text, nullable=True)
answer_context_text: Mapped[str | None] = mapped_column(Text, nullable=True)
card_direction: Mapped[str] = mapped_column(Text, nullable=False)
prompt_modality: Mapped[str] = mapped_column(Text, nullable=False, default="text")
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
default=lambda: datetime.now(timezone.utc),
)
class FlashcardEventEntity(Base):
__tablename__ = "flashcard_event"
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
flashcard_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("flashcard.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
event_type: Mapped[str] = mapped_column(Text, nullable=False)
user_response: Mapped[str | None] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
default=lambda: datetime.now(timezone.utc),
)

View file

@ -15,6 +15,8 @@ from ....domain.models.dictionary import Lemma, Sense, Wordform
class DictionaryRepository(Protocol):
async def get_senses_for_headword(self, headword: str, language: str) -> list[Sense]: ...
async def find_senses_by_english_gloss(self, text: str, target_lang: str) -> list[Sense]: ...
async def get_sense(self, sense_id: uuid.UUID) -> Sense | None: ...
async def get_lemma(self, lemma_id: uuid.UUID) -> Lemma | None: ...
async def get_wordforms_for_lemma(self, lemma_id: uuid.UUID) -> list[Wordform]: ...
@ -29,6 +31,18 @@ def _sense_to_model(entity: DictionarySenseEntity) -> Sense:
)
def _lemma_to_model(entity: DictionaryLemmaEntity) -> Lemma:
return Lemma(
id=str(entity.id),
headword=entity.headword,
language=entity.language,
pos_raw=entity.pos_raw,
pos_normalised=entity.pos_normalised,
gender=entity.gender,
tags=entity.tags or [],
)
def _wordform_to_model(entity: DictionaryWordformEntity) -> Wordform:
return Wordform(
id=str(entity.id),
@ -71,6 +85,20 @@ class PostgresDictionaryRepository:
)
return [_sense_to_model(e) for e in result.scalars().all()]
async def get_sense(self, sense_id: uuid.UUID) -> Sense | None:
result = await self.db.execute(
select(DictionarySenseEntity).where(DictionarySenseEntity.id == sense_id)
)
entity = result.scalar_one_or_none()
return _sense_to_model(entity) if entity else None
async def get_lemma(self, lemma_id: uuid.UUID) -> Lemma | None:
result = await self.db.execute(
select(DictionaryLemmaEntity).where(DictionaryLemmaEntity.id == lemma_id)
)
entity = result.scalar_one_or_none()
return _lemma_to_model(entity) if entity else None
async def get_wordforms_for_lemma(self, lemma_id: uuid.UUID) -> list[Wordform]:
result = await self.db.execute(
select(DictionaryWordformEntity).where(

View file

@ -0,0 +1,136 @@
import uuid
from datetime import datetime, timezone
from typing import Protocol
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from ..entities.flashcard_entities import FlashcardEntity, FlashcardEventEntity
from ....domain.models.flashcard import Flashcard, FlashcardEvent
class FlashcardRepository(Protocol):
async def create_flashcard(
self,
user_id: uuid.UUID,
bank_entry_id: uuid.UUID,
source_lang: str,
target_lang: str,
prompt_text: str,
answer_text: str,
card_direction: str,
prompt_modality: str = "text",
prompt_context_text: str | None = None,
answer_context_text: str | None = None,
) -> Flashcard: ...
async def get_flashcards_for_user(self, user_id: uuid.UUID) -> list[Flashcard]: ...
async def get_flashcards_for_entry(self, bank_entry_id: uuid.UUID) -> list[Flashcard]: ...
async def record_event(
self,
flashcard_id: uuid.UUID,
user_id: uuid.UUID,
event_type: str,
user_response: str | None = None,
) -> FlashcardEvent: ...
def _flashcard_to_model(entity: FlashcardEntity) -> Flashcard:
return Flashcard(
id=str(entity.id),
user_id=str(entity.user_id),
bank_entry_id=str(entity.bank_entry_id),
source_lang=entity.source_lang,
target_lang=entity.target_lang,
prompt_text=entity.prompt_text,
answer_text=entity.answer_text,
prompt_context_text=entity.prompt_context_text,
answer_context_text=entity.answer_context_text,
card_direction=entity.card_direction,
prompt_modality=entity.prompt_modality,
created_at=entity.created_at,
)
def _event_to_model(entity: FlashcardEventEntity) -> FlashcardEvent:
return FlashcardEvent(
id=str(entity.id),
flashcard_id=str(entity.flashcard_id),
user_id=str(entity.user_id),
event_type=entity.event_type,
user_response=entity.user_response,
created_at=entity.created_at,
)
class PostgresFlashcardRepository:
def __init__(self, db: AsyncSession) -> None:
self.db = db
async def create_flashcard(
self,
user_id: uuid.UUID,
bank_entry_id: uuid.UUID,
source_lang: str,
target_lang: str,
prompt_text: str,
answer_text: str,
card_direction: str,
prompt_modality: str = "text",
prompt_context_text: str | None = None,
answer_context_text: str | None = None,
) -> Flashcard:
entity = FlashcardEntity(
user_id=user_id,
bank_entry_id=bank_entry_id,
source_lang=source_lang,
target_lang=target_lang,
prompt_text=prompt_text,
answer_text=answer_text,
prompt_context_text=prompt_context_text,
answer_context_text=answer_context_text,
card_direction=card_direction,
prompt_modality=prompt_modality,
created_at=datetime.now(timezone.utc),
)
self.db.add(entity)
await self.db.commit()
await self.db.refresh(entity)
return _flashcard_to_model(entity)
async def get_flashcards_for_user(self, user_id: uuid.UUID) -> list[Flashcard]:
result = await self.db.execute(
select(FlashcardEntity)
.where(FlashcardEntity.user_id == user_id)
.order_by(FlashcardEntity.created_at.desc())
)
return [_flashcard_to_model(e) for e in result.scalars().all()]
async def get_flashcards_for_entry(self, bank_entry_id: uuid.UUID) -> list[Flashcard]:
result = await self.db.execute(
select(FlashcardEntity)
.where(FlashcardEntity.bank_entry_id == bank_entry_id)
.order_by(FlashcardEntity.created_at.desc())
)
return [_flashcard_to_model(e) for e in result.scalars().all()]
async def record_event(
self,
flashcard_id: uuid.UUID,
user_id: uuid.UUID,
event_type: str,
user_response: str | None = None,
) -> FlashcardEvent:
entity = FlashcardEventEntity(
flashcard_id=flashcard_id,
user_id=user_id,
event_type=event_type,
user_response=user_response,
created_at=datetime.now(timezone.utc),
)
self.db.add(entity)
await self.db.commit()
await self.db.refresh(entity)
return _event_to_model(entity)

View file

@ -37,6 +37,8 @@ class VocabRepository(Protocol):
self, entry_id: uuid.UUID, sense_id: uuid.UUID
) -> LearnableWordBankEntry: ...
async def get_entry(self, entry_id: uuid.UUID) -> LearnableWordBankEntry | None: ...
async def get_pending_disambiguation(self, user_id: uuid.UUID) -> list[LearnableWordBankEntry]: ...
@ -154,6 +156,15 @@ class PostgresVocabRepository:
await self.db.refresh(entity)
return _entry_to_model(entity)
async def get_entry(self, entry_id: uuid.UUID) -> LearnableWordBankEntry | None:
result = await self.db.execute(
select(LearnableWordBankEntryEntity).where(
LearnableWordBankEntryEntity.id == entry_id
)
)
entity = result.scalar_one_or_none()
return _entry_to_model(entity) if entity else None
async def get_pending_disambiguation(self, user_id: uuid.UUID) -> list[LearnableWordBankEntry]:
result = await self.db.execute(
select(LearnableWordBankEntryEntity)

View file

@ -0,0 +1,143 @@
import uuid
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncSession
from ...auth import verify_token
from ...domain.services.flashcard_service import FlashcardService
from ...outbound.postgres.database import get_db
from ...outbound.postgres.repositories.dictionary_repository import PostgresDictionaryRepository
from ...outbound.postgres.repositories.flashcard_repository import PostgresFlashcardRepository
from ...outbound.postgres.repositories.vocab_repository import PostgresVocabRepository
router = APIRouter(tags=["flashcards"])
class FlashcardResponse(BaseModel):
id: str
user_id: str
bank_entry_id: str
source_lang: str
target_lang: str
prompt_text: str
answer_text: str
prompt_context_text: str | None
answer_context_text: str | None
card_direction: str
prompt_modality: str
created_at: str
class FlashcardEventResponse(BaseModel):
id: str
flashcard_id: str
user_id: str
event_type: str
user_response: str | None
created_at: str
class GenerateFlashcardsRequest(BaseModel):
direction: str | None = None
class RecordEventRequest(BaseModel):
event_type: str
user_response: str | None = None
def _service(db: AsyncSession) -> FlashcardService:
return FlashcardService(
flashcard_repo=PostgresFlashcardRepository(db),
vocab_repo=PostgresVocabRepository(db),
dict_repo=PostgresDictionaryRepository(db),
)
@router.post(
"/vocab/{entry_id}/flashcards",
response_model=list[FlashcardResponse],
status_code=status.HTTP_201_CREATED,
)
async def generate_flashcards(
entry_id: str,
body: GenerateFlashcardsRequest,
db: AsyncSession = Depends(get_db),
token_data: dict = Depends(verify_token),
) -> list[FlashcardResponse]:
try:
eid = uuid.UUID(entry_id)
except ValueError:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid entry_id")
try:
cards = await _service(db).generate_flashcard_from_entry(eid, direction=body.direction)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc))
return [_flashcard_response(c) for c in cards]
@router.get("/flashcards", response_model=list[FlashcardResponse])
async def list_flashcards(
db: AsyncSession = Depends(get_db),
token_data: dict = Depends(verify_token),
) -> list[FlashcardResponse]:
user_id = uuid.UUID(token_data["sub"])
cards = await PostgresFlashcardRepository(db).get_flashcards_for_user(user_id)
return [_flashcard_response(c) for c in cards]
@router.post(
"/flashcards/{flashcard_id}/events",
response_model=FlashcardEventResponse,
status_code=status.HTTP_201_CREATED,
)
async def record_event(
flashcard_id: str,
body: RecordEventRequest,
db: AsyncSession = Depends(get_db),
token_data: dict = Depends(verify_token),
) -> FlashcardEventResponse:
try:
fid = uuid.UUID(flashcard_id)
except ValueError:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid flashcard_id")
user_id = uuid.UUID(token_data["sub"])
try:
event = await _service(db).record_flashcard_event(
flashcard_id=fid,
user_id=user_id,
event_type=body.event_type,
response=body.user_response,
)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))
return FlashcardEventResponse(
id=event.id,
flashcard_id=event.flashcard_id,
user_id=event.user_id,
event_type=event.event_type,
user_response=event.user_response,
created_at=event.created_at.isoformat(),
)
def _flashcard_response(card) -> FlashcardResponse:
return FlashcardResponse(
id=card.id,
user_id=card.user_id,
bank_entry_id=card.bank_entry_id,
source_lang=card.source_lang,
target_lang=card.target_lang,
prompt_text=card.prompt_text,
answer_text=card.answer_text,
prompt_context_text=card.prompt_context_text,
answer_context_text=card.answer_context_text,
card_direction=card.card_direction,
prompt_modality=card.prompt_modality,
created_at=card.created_at.isoformat(),
)

View file

@ -1,4 +1,5 @@
from .account import router as account_router
from .flashcards import router as flashcards_router
from .pos import router as pos_router
from .translate import router as translate_router
from .generation import router as generation_router
@ -11,6 +12,7 @@ from fastapi import APIRouter
api_router = APIRouter(prefix="/api", tags=["api"])
api_router.include_router(account_router)
api_router.include_router(flashcards_router)
api_router.include_router(pos_router)
api_router.include_router(translate_router)
api_router.include_router(generation_router)