Build Call and Task Models for VitalAI
Author SQLAlchemy 2.0 models and fix Alembic PostgreSQL enum migrations.
Introduction
30 Second Summary
Healthcare intake systems handle dozens of phone calls and task assignments every day. Each interaction needs a structured record that connects back to the patient's case, tracks its status, and routes it to the right person.
In this project, you will author two SQLAlchemy 2.0 data models for VitalAI's intake pipeline, generate an Alembic migration, and fix the well-documented PostgreSQL enum downgrade bug before it breaks your team's migration chain.
What You'll Build
You'll run the full pytest suite and watch it pass against real Postgres, proving your new Call and Task models are wired correctly into the same migration pipeline your team uses in production.
By the end of this project, you'll have:
- Two production-ready models (Call and Task) using SQLAlchemy 2.0's Mapped/mapped_column declarative style with StrEnum-backed PostgreSQL native enums.
- A reviewed and fixed Alembic migration with a manually-corrected downgrade that drops enum types. The autogenerate silently skips this step.
- Confidence to explain every pattern choice (values_callable, explicit enum naming, FK indexing) in your subject coordinator Q&A.
- Secret Mission: Verify the full downgrade-upgrade cycle works end-to-end and add a partial index for efficient task queue lookups.
Are there any prerequisites?
This project assumes you have the existing VitalAI repo with Docker Compose, Alembic, an async engine, and the base models (IntakeCase, ConsentRecord, User, Appointment) already configured and migrated from Sprints 1-2.
You'll also need Python 3.13 and a working Docker setup on your machine.
Before We Start
Before we dive in, let's lock in what you're building for Sprint 3 and why the intake pipeline needs dedicated models for phone calls and routed tasks.
Author the Call and Task Models
The VitalAI processing pipeline needs dedicated records for phone calls and routed tasks. Without these models, incoming calls have nowhere to land and classified work items cannot be tracked through staff queues.
In this step, you will create two new model files that follow the exact conventions your team established in Sprints 1-2. Each model uses SQLAlchemy 2.0's Mapped/mapped_column declarative style with StrEnum, explicit PostgreSQL enum naming, and indexed foreign keys.
In this step, get ready to:
- Create the Call model with a CallStatus enum for tracking phone interactions.
- Create the Task model with three enums (TaskSource, TaskPriority, TaskStatus) for routing work items.
- Verify both models import cleanly against your existing project.
Create the Call model
Every phone interaction in VitalAI's intake pipeline needs a persistent record. The Call model stores the caller's number, an optional transcript, and a status that tracks whether the call has been processed or escalated.
- Create a new file at app/models/call.py and add the following code:
import enum
from uuid import UUID
from sqlalchemy import Enum, ForeignKey, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base, TimestampMixin, UUIDPrimaryKeyMixin
class CallStatus(enum.StrEnum):
RECEIVED = "received"
PROCESSED = "processed"
ESCALATED = "escalated"
class Call(Base, UUIDPrimaryKeyMixin, TimestampMixin):
__tablename__ = "calls"
case_id: Mapped[UUID] = mapped_column(
ForeignKey("intake_cases.id", ondelete="CASCADE"), nullable=False, index=True
)
phone_number: Mapped[str] = mapped_column(String(20), nullable=False)
transcript: Mapped[str | None] = mapped_column(Text, nullable=True)
status: Mapped[CallStatus] = mapped_column(
Enum(CallStatus, name="call_status", values_callable=lambda x: [e.value for e in x]),
nullable=False,
default=CallStatus.RECEIVED,
)
What does this code do?
CallStatus is a StrEnum, which means each member IS its string value (CallStatus.RECEIVED == "received"). This makes comparisons natural in both Python and raw SQL.
The Enum() column type uses two critical arguments:
- name="call_status" tells PostgreSQL to create a CREATE TYPE call_status instead of using the Python class name.
- values_callable=lambda x: [e.value for e in x] persists the lowercase values ("received") to the database instead of the uppercase Python names (RECEIVED).
- case_id uses index=True because queries will frequently filter calls by case. The ondelete="CASCADE" ensures that deleting a case removes its calls automatically.
- Save app/models/call.py.
- Confirm the model imports cleanly by running:
python -c "from app.models.call import Call, CallStatus; print('Call model OK')"
You should see Call model OK printed with no import errors.
Seeing a ModuleNotFoundError?
- Check that you created the file at the exact path app/models/call.py. A typo in the directory name will prevent Python from finding the module.
- Make sure your terminal's working directory is the project root (the folder containing the app/ directory).
- Verify that app/models/base.py exists and exports Base, TimestampMixin, and UUIDPrimaryKeyMixin. These were created in an earlier sprint.
Why can't Python find my Call model?
Create the Task model
Tasks represent work items routed to staff queues after a call is classified. The Task model needs three enums to capture where the task came from, how urgent it is, and where it sits in the workflow.
- Create a new file at app/models/task.py and add the imports and enum definitions:
import enum
from uuid import UUID
from sqlalchemy import Enum, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base, TimestampMixin, UUIDPrimaryKeyMixin
class TaskSource(enum.StrEnum):
EMAIL = "email"
CALL = "call"
class TaskPriority(enum.StrEnum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
URGENT = "urgent"
class TaskStatus(enum.StrEnum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
ESCALATED = "escalated"
Why three separate enums?
Each enum maps to a distinct PostgreSQL CREATE TYPE. Keeping them separate means you can add a new source (like "chat") without touching priority or status definitions.
Notice that TaskStatus shares the ESCALATED value with CallStatus. They live in separate PostgreSQL types (task_status vs call_status) so there is no collision.
- Below the TaskStatus class in the same file, add the Task model class:
class Task(Base, UUIDPrimaryKeyMixin, TimestampMixin):
__tablename__ = "tasks"
case_id: Mapped[UUID] = mapped_column(
ForeignKey("intake_cases.id", ondelete="CASCADE"), nullable=False, index=True
)
assigned_to: Mapped[UUID | None] = mapped_column(
ForeignKey("users.id"), nullable=True, index=True
)
source: Mapped[TaskSource] = mapped_column(
Enum(TaskSource, name="task_source", values_callable=lambda x: [e.value for e in x]),
nullable=False,
)
priority: Mapped[TaskPriority] = mapped_column(
Enum(TaskPriority, name="task_priority", values_callable=lambda x: [e.value for e in x]),
nullable=False,
default=TaskPriority.MEDIUM,
)
status: Mapped[TaskStatus] = mapped_column(
Enum(TaskStatus, name="task_status", values_callable=lambda x: [e.value for e in x]),
nullable=False,
default=TaskStatus.PENDING,
)
What does this model do?
- case_id links back to the intake case with CASCADE delete, same pattern as the Call model.
- assigned_to is a nullable FK to users.id. Unassigned tasks sit in the queue until a staff member claims them. The FK deliberately omits ondelete because user-deletion behavior is a team decision for a future sprint.
- Each enum column uses name= to set an explicit PostgreSQL type name (task_source, task_priority, task_status) rather than letting SQLAlchemy default to the Python class name in lowercase.
- priority defaults to TaskPriority.MEDIUM and status defaults to TaskStatus.PENDING, matching the team's append-only enum conventions.
- Save app/models/task.py.
✔️ Awesome, I've got everything!
Great. Make sure you have saved both app/models/call.py and app/models/task.py.
ⓧ I'd like to double check the full code
Your app/models/call.py should match this exactly:
import enum
from uuid import UUID
from sqlalchemy import Enum, ForeignKey, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base, TimestampMixin, UUIDPrimaryKeyMixin
class CallStatus(enum.StrEnum):
RECEIVED = "received"
PROCESSED = "processed"
ESCALATED = "escalated"
class Call(Base, UUIDPrimaryKeyMixin, TimestampMixin):
__tablename__ = "calls"
case_id: Mapped[UUID] = mapped_column(
ForeignKey("intake_cases.id", ondelete="CASCADE"), nullable=False, index=True
)
phone_number: Mapped[str] = mapped_column(String(20), nullable=False)
transcript: Mapped[str | None] = mapped_column(Text, nullable=True)
status: Mapped[CallStatus] = mapped_column(
Enum(CallStatus, name="call_status", values_callable=lambda x: [e.value for e in x]),
nullable=False,
default=CallStatus.RECEIVED,
)
For app/models/task.py, scroll up and confirm you have both blocks in order: the imports and three StrEnum classes at the top, followed by the Task class below them. The full file is 52 lines. Verify the enums are TaskSource, TaskPriority, TaskStatus (in that order), and the model class has five columns: case_id, assigned_to, source, priority, status.
Verify both models import cleanly
Before you run this, what do you expect to happen? If both files are saved correctly with matching imports, you should see a success message.
- Test that both models resolve against your existing codebase by running:
python -c "from app.models.call import Call, CallStatus; from app.models.task import Task, TaskSource, TaskPriority, TaskStatus; print('All models import cleanly')"
You should see All models import cleanly printed with no errors.
Seeing an ImportError or NameError?
- If you see ImportError: cannot import name 'CallStatus', check that the class name in your file matches exactly (capital C, capital S).
- A NameError inside the model definition usually means a typo in the Enum() arguments. Verify that values_callable is spelled correctly and uses a lambda, not a plain function reference.
- Make sure both files import from app.models.base (not a relative import like .base).
Help me debug my model import error.
Both models are ready. Next up, you will register them in the package __init__.py so Alembic's autogenerate can detect them and build your migration.
Register Models and Generate the Migration
Your Call and Task models are defined and import cleanly. Now it's time to wire them into Alembic so the migration pipeline actually sees them.
Alembic's autogenerate works by comparing the live database schema against your Base.metadata object. If the new models aren't imported somewhere that touches that metadata, autogenerate produces an empty migration with no table changes detected.
In this step, get ready to:
- Register Call and Task in app/models/__init__.py.
- Generate the migration with Alembic autogenerate.
- Review the upgrade() function to confirm correct table and enum creation.
Register the new models in __init__.py
Your package's __init__.py is what ties model classes to Base.metadata. Without these imports, Python never executes the model class bodies, and SQLAlchemy never registers the tables on the metadata.
- Open app/models/__init__.py and add these import lines alongside the existing model imports:
from app.models.call import Call, CallStatus
from app.models.task import Task, TaskPriority, TaskSource, TaskStatus
- Find the existing __all__ list in the same file and add the six new names to it:
"Call", "CallStatus", "Task", "TaskPriority", "TaskSource", "TaskStatus"
Why does __all__ matter here?
The __all__ list controls what gets exported when another module does from app.models import *. Adding your new classes here keeps the public API explicit and consistent with your team's existing pattern for IntakeCase, ConsentRecord, and the other models.
- Save app/models/__init__.py.
- Confirm the package exports work by running:
python -c "from app.models import Call, Task; print('Package exports OK')"
You should see Package exports OK printed with no import errors.
Seeing an ImportError?
- Check that the import paths match your file names exactly. from app.models.call import Call requires a file named call.py inside app/models/.
- Verify you're running the command from the project root (the directory containing app/).
- Help me debug this import error
Generate the migration with autogenerate
With both models registered on the metadata, Alembic can now diff the live Postgres schema against what your Python models describe. The result is an autogenerated migration script containing the exact DDL needed to bring the database in sync.
Before you run this, what do you expect Alembic to detect? Think about which tables and which PostgreSQL types it needs to create.
- Generate the migration by running:
alembic revision --autogenerate -m "add_calls_and_tasks_tables"
You should see output containing Generating ... add_calls_and_tasks_tables.py ... done with detected changes for both tables.
What does autogenerate actually compare?
Alembic connects to your running Postgres instance and reads the current schema. It then compares that against every table registered on Base.metadata in your Python code.
Any model that is imported (and therefore has its class body executed) gets registered on the metadata. That's why the __init__.py step was critical. Without it, Call and Task would be invisible to autogenerate.
Migration file is empty or shows no changes?
- Confirm your __init__.py imports are saved. Unsaved changes mean Python still can't see the models.
- Check that your Alembic env.py imports from app.models so that all registered models are loaded before autogenerate runs.
- Make sure your Docker Postgres is running. Alembic needs to connect to produce the diff.
- Help me figure out why autogenerate detected no changes
Review the generated upgrade() function
Autogenerated migrations are a starting point, not a finished product. Before you move on, read through the migration to confirm Alembic interpreted your models correctly.
- Open the newly generated file inside alembic/versions/. Its filename ends with _add_calls_and_tasks_tables.py.
- Scroll to the upgrade() function.
Confirm you see the following in upgrade():
- An op.create_table("calls", ...) call containing a call_status enum column, a case_id foreign key to intake_cases.id, phone_number, and transcript columns.
- An op.create_table("tasks", ...) call containing task_source, task_priority, and task_status enum columns, a case_id FK to intake_cases.id, and a nullable assigned_to FK to users.id.
- Enum values rendered as lowercase strings (received, processed, escalated, etc.) rather than Python names (RECEIVED, PROCESSED). This confirms values_callable worked.
What about the downgrade() function?
Now scroll down to downgrade(). You'll see it contains only op.drop_table("tasks") and op.drop_table("calls").
Notice anything missing? The upgrade creates four PostgreSQL enum types (call_status, task_source, task_priority, task_status), but the downgrade never drops them. That's a real bug you'll fix in the next step.
Your migration is generated and its upgrade path looks correct. Next up, you'll fix the downgrade function to handle PostgreSQL's enum types properly before this migration is safe to merge.
Review and Fix the Migration's Downgrade
Your migration file exists and its upgrade() function looks correct. Before you apply it, there's a critical review step that separates a junior contributor from someone ready for coordinator Q&A.
Alembic's autogenerate has a well-documented limitation with PostgreSQL native enum types. It creates them on upgrade but silently leaves them behind on downgrade. If you merge this migration unfixed, any teammate who runs a downgrade and re-upgrade will hit "type already exists" errors that break the entire migration chain.
In this step, get ready to:
- Inspect the autogenerated downgrade and identify the missing enum cleanup.
- Add explicit DROP TYPE IF EXISTS statements for all four enum types.
- Verify the upgrade has correct enum names and values_callable-rendered values.
Inspect the autogenerated downgrade
Every Alembic migration has two functions: upgrade() applies the change, and downgrade() reverses it. For the migration to be safe, downgrade must undo everything upgrade created. This includes not just the tables but also the PostgreSQL enum types that the tables reference.
- Open your migration file in alembic/versions/. It's the file ending in _add_calls_and_tasks_tables.py.
- Scroll down to the downgrade() function at the bottom of the file.
Before you read it, ask yourself: your upgrade creates two tables and four PostgreSQL enum types. What should the downgrade contain to fully reverse that?
Now read the function. You'll see something like this:
def downgrade() -> None:
op.drop_table('tasks')
op.drop_table('calls')
That's it. Two table drops and nothing else. The four PostgreSQL enum types (call_status, task_source, task_priority, task_status) that upgrade() created via CREATE TYPE are left orphaned in the database.
Why does Alembic leave the enum types behind?
This is a long-standing limitation in Alembic's autogenerate. When sa.Enum is embedded inside op.create_table, the upgrade implicitly issues CREATE TYPE before creating the table. But the downgrade logic only knows about the table itself, not the dependent types.
The Alembic maintainer has confirmed this as a known limitation (GitHub issues #67, #278, #754, #886). The recommended fix is to manually drop enums in the downgrade function.
Add the DROP TYPE IF EXISTS statements
You need to add explicit SQL statements that drop each enum type after the tables are gone. The tables must be dropped first because they reference the types.
- In your migration file, find the downgrade() function. It currently looks like this:
def downgrade() -> None:
op.drop_table('tasks')
op.drop_table('calls')
- Replace the entire downgrade() function with the fixed version below:
def downgrade() -> None:
# Drop tables first (they reference the enum types)
op.drop_table('tasks')
op.drop_table('calls')
# Manually drop the PostgreSQL enum types that autogenerate missed
op.execute(sa.text('DROP TYPE IF EXISTS call_status'))
op.execute(sa.text('DROP TYPE IF EXISTS task_source'))
op.execute(sa.text('DROP TYPE IF EXISTS task_priority'))
op.execute(sa.text('DROP TYPE IF EXISTS task_status'))
What does this fix do?
- op.execute(sa.text(...)) runs raw SQL through Alembic's connection. You need raw SQL because Alembic has no built-in operation for dropping enum types.
- IF EXISTS makes each statement idempotent. If a partial downgrade already removed some types, re-running won't error out.
- The order matters: tables drop first because their columns have a dependency on the enum types. Dropping a type that's still referenced by a column would fail.
- Save the migration file.
Not sure where sa.text comes from?
Alembic's autogenerated migration files include import sqlalchemy as sa at the top. Check that this import exists in your file. If it's missing for any reason, add it below the other imports.
Help me understand why sa.text is needed in my migration
Verify the upgrade has correct enum names and values
While you have the migration open, scroll back up to the upgrade() function. You need to confirm two things before this migration is safe to run.
- Check that the sa.Enum calls inside each op.create_table use the correct type names: call_status, task_source, task_priority, and task_status. These must match the name= parameters in your model definitions.
- Verify that values_callable rendered the lowercase string values into the migration. You should see the actual values ('received', 'processed', 'escalated', etc.) in the enum definitions, not the Python member names ('RECEIVED', 'PROCESSED', etc.).
Why do the enum values matter so much?
PostgreSQL stores the enum values exactly as they appear in CREATE TYPE. If the migration stores 'RECEIVED' but your application writes 'received', every insert will fail with a type mismatch. The values_callable parameter ensures the lowercase .value strings from your StrEnum are what gets persisted to PostgreSQL.
If both checks pass, your migration is ready. The downgrade() now drops both tables then all four enum types. The upgrade() creates the types with correct lowercase values and builds both tables referencing them.
Your migration is reviewed, fixed, and ready to run against real Postgres. Next up, you'll apply it, verify the schema directly in the database, and prove the full test suite still passes.
Apply the Migration and Run Tests
Your migration file is reviewed, fixed, and ready. But a migration only becomes real when it runs against the actual database engine your production environment uses.
Schema changes must be validated against real PostgreSQL because native enums, CREATE TYPE, and FK constraints behave differently than SQLite. Running the full test suite afterwards confirms your new models don't break any existing IntakeCase, ConsentRecord, or Appointment tests.
In this step, get ready to:
- Apply the migration to your Dockerized Postgres with alembic upgrade head.
- Verify the schema directly using psql meta-commands.
- Run the full pytest suite and confirm zero failures.
Apply the migration to Postgres
Your Dockerized Postgres needs to be running before Alembic can connect and apply the migration.
- Ensure your database container is running by executing:
docker compose up -d db
You should see the container listed as running (or already up if it was started earlier).
Before you run the migration, what do you expect to see if everything works? Think about what Alembic prints when it applies a revision successfully.
- Apply the migration by running:
alembic upgrade head
You should see output containing Running upgrade ... -> ..., add_calls_and_tasks_tables completing without error. That single line confirms Alembic connected to Postgres, issued the CREATE TYPE statements for your four enums, and created both tables with all their constraints.
Seeing a connection error?
Make sure your .env or alembic.ini has the correct database URL pointing to your Docker Postgres instance. Confirm the container is healthy with docker compose ps.
If you see "type already exists", it means a previous partial run left orphaned enum types. Connect to psql and manually drop them, then retry the upgrade.
Help me debug my Alembic connection to Dockerized Postgres.
Verify the schema directly in psql
Alembic reported success, but you should always verify the actual database state. Connecting directly with psql lets you confirm the tables, enum types, columns, and indexes all exist as expected.
- Connect to your Postgres container by running:
docker compose exec db psql -U [[DB_USER="your postgres username"]] -d [[DB_NAME="your database name"]]
- List all tables to confirm calls and tasks appear by running:
\dt
You should see calls and tasks listed alongside your existing tables (intake_cases, users, etc.).
- List all custom enum types to confirm your four new types exist:
\dT+
You should see call_status, task_source, task_priority, and task_status in the output with their element values listed.
- Inspect the calls table structure:
\d calls
Confirm you see columns for id, case_id, phone_number, transcript, status, created_at, and updated_at. You should also see an index on case_id.
- Inspect the tasks table structure:
\d tasks
Confirm columns for id, case_id, assigned_to, source, priority, status, created_at, and updated_at. You should see indexes on both case_id and assigned_to.
- Exit psql by typing:
\q
Why verify manually when Alembic said it worked?
Alembic reports success if the Python code executed without raising an exception. It does not verify that the resulting schema matches your model definitions. A typo in an enum name or a missing index would still produce a "successful" migration.
Checking with psql gives you ground truth. You can confirm column types, nullable flags, FK targets, and index presence directly against the database catalog.
Run the full test suite
The schema looks correct in psql. Now you need to confirm your new models don't introduce regressions against the existing test suite that covers IntakeCase, ConsentRecord, and Appointment.
Before you run the tests, what do you expect the result to be? Your new models add tables but don't modify any existing ones. Think about whether that should cause any test failures.
- Run the full test suite:
pytest
You should see the entire suite pass with zero failures and zero errors. Every existing test for IntakeCase, ConsentRecord, User, and Appointment continues to pass because your new models add to the schema without modifying existing tables.
Seeing test failures?
If you see import errors, check that app/models/__init__.py has the correct import paths for Call, CallStatus, Task, TaskSource, TaskPriority, and TaskStatus.
If you see metadata conflicts, ensure your model class names and __tablename__ values don't collide with existing models. The table names should be calls and tasks.
Help me debug my pytest failures after adding new SQLAlchemy models.
Secret mission
Verify the Downgrade Cycle and Add a Queue Index
Your downgrade fix from Step 3 is only proven correct when you actually run the full cycle against real Postgres. Go further by writing a second migration that adds a partial index for the task queue pattern, giving unassigned tasks an efficient lookup path for next sprint's routing logic.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely on local Docker, so there are no ongoing costs.
Resources you used:
- Two PostgreSQL tables (calls, tasks) and four enum types in your Dockerized Postgres.
- Partial index tasks_unassigned_queue_idx on the tasks table.
- Two Alembic migration files in your alembic/versions/ directory.
- Docker Compose Postgres service and its data volume.
Keep everything running
No action needed. Choose this if you are still actively building on the VitalAI repo during your sprint work.
- Leave Docker Compose running. Your Postgres container retains all tables, enum types, and indexes.
- Your two model files and two migration files remain in the repo, ready for the next sprint.
Pause - I'll come back to this later
Shut down the running containers to free up system memory without losing any data.
- Stop Docker Compose by running:
docker compose stop
Your database volume is preserved on disk. When you are ready to resume, run:
docker compose up -d db
All tables, enum types, and the partial index will still be there.
Delete - I don't want to use this again
Remove all database state and Docker resources completely. Only choose this if you want to start fresh.
- Roll back all migrations to remove the tables, enum types, and index from the database by running:
alembic downgrade base
- Tear down the Docker Compose services and wipe the Postgres data volume by running:
docker compose down -v
- Delete the two migration files from your versions directory by running:
rm alembic/versions/*_add_calls_and_tasks_tables.py
rm alembic/versions/*_add_tasks_unassigned_queue_index.py
What about the model files?
The files app/models/call.py and app/models/task.py are source code, not ephemeral resources. Delete them only if you truly want to remove all Sprint 3 work from the repo.
Nice Work!
Nice Work!
You just built two production-ready data models and a reviewed, fixed migration for VitalAI's intake pipeline. Your Call and Task models are live in Postgres, backed by a test suite that proves zero regressions.
You've learned how to:
- Author SQLAlchemy 2.0 models using the Mapped/mapped_column declarative style with StrEnum, values_callable, and explicit PostgreSQL enum naming.
- Catch and fix the well-documented Alembic PostgreSQL enum downgrade bug where DROP TYPE is not auto-generated, before it breaks the migration chain for your team.
- Apply and verify migrations against real Dockerized Postgres, confirming correct schema creation and zero test regressions across all existing models.
- Secret Mission: Proved the downgrade-upgrade cycle works end-to-end and added a partial index for efficient task queue lookups.
Ready to quiz yourself?