Introduction

The previous article created the FastAPI project structure for the AI Tender Opportunity Scanner.

The backend now includes:

A working FastAPI application
Environment-based configuration
A PostgreSQL connection
A SQLAlchemy declarative base
Versioned API routing
A health-check endpoint
Initial test infrastructure
Directories for models, schemas, repositories, services, integrations, and matching logic

The next step is translating the database design into executable SQLAlchemy models.

These models will represent the main entities in the application:

Users
Business profiles
Tender sources
Ingestion runs
Raw tender records
Normalized tenders
Tender summaries
Tender embeddings
Business profile embeddings
Tender matches
Saved tenders
Dismissed tenders
Processing errors

We will use the modern SQLAlchemy declarative mapping style with Mapped and mapped_column. This provides typed model attributes while keeping table definitions and relationships inside normal Python classes.

By the end of this article, the complete MVP database model layer will be ready for Alembic migrations.

Development Progress

			
Project Scope and Roadmap        ████████████████████ 100%
MVP Definition                   ████████████████████ 100%
Technology Stack                 ████████████████████ 100%
System Architecture              ████████████████████ 100%
Database Design                  ████████████████████ 100%
Backend Project Structure        ████████████████████ 100%
SQLAlchemy Models                ████████████████████ 100%
Alembic Migrations               ░░░░░░░░░░░░░░░░░░░░   0%
Tender Ingestion                 ░░░░░░░░░░░░░░░░░░░░   0%
AI Analysis                      ░░░░░░░░░░░░░░░░░░░░   0%
Matching Engine                  ░░░░░░░░░░░░░░░░░░░░   0%
Frontend Dashboard               ░░░░░░░░░░░░░░░░░░░░   0%

		

What We Are Building

The model layer will define:

PostgreSQL table names
UUID primary keys
Column types
Required and optional fields
Foreign keys
One-to-one relationships
One-to-many relationships
Unique constraints
Check constraints
Database indexes
JSONB fields
Vector fields
Timestamp fields
Deletion behavior

We will not create the database tables directly in this article.

The models describe the desired database structure. Alembic will turn that structure into migration scripts in the next article.

Updated Model Directory

The app/models directory will contain:

			
app/models/
├── __init__.py
├── mixins.py
├── user.py
├── business_profile.py
├── tender_source.py
├── ingestion.py
├── tender.py
├── tender_summary.py
├── embedding.py
├── tender_match.py
├── user_tender_action.py
└── processing_error.py

		

Separating the models by domain keeps individual files manageable.

Step 1: Install pgvector Support

The database design includes vector embeddings for tenders and business profiles.

Install the Python pgvector package:

pip install pgvector

Update the dependency snapshot:

pip freeze > requirements.txt

The pgvector Python package provides SQLAlchemy support through its VECTOR column type. PostgreSQL must also have the pgvector extension installed and enabled before vector columns can be created.

We will enable the PostgreSQL extension through an Alembic migration in the next article.

Step 2: Improve the Declarative Base

The previous article created a minimal SQLAlchemy base class.

Open:

app/database/base.py

Replace its contents with:

			
from sqlalchemy import MetaData
from sqlalchemy.orm import DeclarativeBase
NAMING_CONVENTION = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": (
        "fk_%(table_name)s_%(column_0_name)s_"
        "%(referred_table_name)s"
    ),
    "pk": "pk_%(table_name)s",
}
class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=NAMING_CONVENTION)

		

Why Add a Naming Convention?

Without a naming convention, PostgreSQL or SQLAlchemy may generate database-specific names for constraints.

That makes migrations harder to understand and can complicate future constraint removal.

The convention gives predictable names to:

Primary keys
Foreign keys
Unique constraints
Check constraints
Indexes

Alembic specifically recommends consistent constraint naming for migration workflows.

For example:

			
pk_users
fk_business_profiles_user_id_users
uq_users_email
ck_tender_matches_overall_score_range

Step 3: Create Shared Model Mixins

Several models need the same fields:

UUID primary key
Creation timestamp
Update timestamp

Create:

app/models/mixins.py

Add:

			
from datetime import datetime
from uuid import UUID, uuid4
from sqlalchemy import DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
class UUIDPrimaryKeyMixin:
    id: Mapped[UUID] = mapped_column(
        primary_key=True,
        default=uuid4,
    )
class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),
        nullable=False,
    )

		

Application Defaults and Server Defaults

The UUID is generated by Python using:

default=uuid4

Timestamps use database-side defaults:

server_default=func.now()

This means PostgreSQL can populate the timestamp even when a row is inserted outside the FastAPI application.

The onupdate option updates the ORM value when SQLAlchemy performs an update. We will still review timestamp behavior when testing the migrations and repositories.

Why Use UUID Primary Keys?

UUIDs are appropriate for this application because they:

Avoid exposing sequential record counts
Can be generated before inserting a row
Work well across distributed components
Reduce collision risks during imports or data merging
Are suitable for public API identifiers

SQLAlchemy supports database-native UUID values and typed UUID mapping.

Step 4: Create the User Model

Create:

app/models/user.py

Add:

			
from __future__ import annotations
from typing import TYPE_CHECKING
from sqlalchemy import Boolean, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.base import Base
from app.models.mixins import TimestampMixin, UUIDPrimaryKeyMixin
if TYPE_CHECKING:
    from app.models.business_profile import BusinessProfile
    from app.models.user_tender_action import (
        DismissedTender,
        SavedTender,
    )
class User(UUIDPrimaryKeyMixin, TimestampMixin, Base):
    __tablename__ = "users"
    email: Mapped[str] = mapped_column(
        String(320),
        nullable=False,
        unique=True,
        index=True,
    )
    hashed_password: Mapped[str | None] = mapped_column(
        String(255),
        nullable=True,
    )
    full_name: Mapped[str | None] = mapped_column(
        String(255),
        nullable=True,
    )
    is_active: Mapped[bool] = mapped_column(
        Boolean,
        default=True,
        server_default="true",
        nullable=False,
    )
    is_admin: Mapped[bool] = mapped_column(
        Boolean,
        default=False,
        server_default="false",
        nullable=False,
    )
    business_profiles: Mapped[list[BusinessProfile]] = relationship(
        back_populates="user",
        cascade="all, delete-orphan",
        passive_deletes=True,
    )
    saved_tenders: Mapped[list[SavedTender]] = relationship(
        back_populates="user",
        cascade="all, delete-orphan",
        passive_deletes=True,
    )
    dismissed_tenders: Mapped[list[DismissedTender]] = relationship(
        back_populates="user",
        cascade="all, delete-orphan",
        passive_deletes=True,
    )

		

Understanding the User Relationships

A user can have:

			
One user -> many business profiles
One user -> many saved tenders
One user -> many dismissed tenders

The MVP interface may initially expose only one active business profile, but the relational model supports multiple profiles.

The relationship configuration includes:

cascade="all, delete-orphan"

This tells SQLAlchemy that child records belong to the parent relationship.

It also includes:

passive_deletes=True

This allows PostgreSQL foreign-key cascade rules to handle deletion without SQLAlchemy first loading every related record.

Step 5: Create the Business Profile Model

Create:

app/models/business_profile.py

Add:

			
from __future__ import annotations
from decimal import Decimal
from typing import TYPE_CHECKING
from uuid import UUID
from sqlalchemy import (
    Boolean,
    CheckConstraint,
    ForeignKey,
    Index,
    Numeric,
    String,
    Text,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.base import Base
from app.models.mixins import TimestampMixin, UUIDPrimaryKeyMixin
if TYPE_CHECKING:
    from app.models.embedding import BusinessProfileEmbedding
    from app.models.processing_error import ProcessingError
    from app.models.tender_match import TenderMatch
    from app.models.user import User
    from app.models.user_tender_action import (
        DismissedTender,
        SavedTender,
    )
class BusinessProfile(
    UUIDPrimaryKeyMixin,
    TimestampMixin,
    Base,
):
    __tablename__ = "business_profiles"
    __table_args__ = (
        CheckConstraint(
            "minimum_contract_value IS NULL "
            "OR minimum_contract_value >= 0",
            name="minimum_contract_value_non_negative",
        ),
        CheckConstraint(
            "maximum_contract_value IS NULL "
            "OR maximum_contract_value >= 0",
            name="maximum_contract_value_non_negative",
        ),
        CheckConstraint(
            "minimum_contract_value IS NULL "
            "OR maximum_contract_value IS NULL "
            "OR minimum_contract_value <= maximum_contract_value",
            name="contract_value_range_valid",
        ),
        Index(
            "ix_business_profiles_user_active",
            "user_id",
            "is_active",
        ),
    )
    user_id: Mapped[UUID] = mapped_column(
        ForeignKey(
            "users.id",
            ondelete="CASCADE",
        ),
        nullable=False,
        index=True,
    )
    name: Mapped[str] = mapped_column(
        String(150),
        nullable=False,
    )
    company_name: Mapped[str | None] = mapped_column(
        String(255),
        nullable=True,
    )
    company_description: Mapped[str] = mapped_column(
        Text,
        nullable=False,
    )
    services: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    target_industries: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    preferred_countries: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    preferred_regions: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    included_keywords: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    excluded_keywords: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    minimum_contract_value: Mapped[Decimal | None] = mapped_column(
        Numeric(18, 2),
        nullable=True,
    )
    maximum_contract_value: Mapped[Decimal | None] = mapped_column(
        Numeric(18, 2),
        nullable=True,
    )
    preferred_currency: Mapped[str | None] = mapped_column(
        String(3),
        nullable=True,
    )
    certifications: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    preferred_languages: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    is_active: Mapped[bool] = mapped_column(
        Boolean,
        default=True,
        server_default="true",
        nullable=False,
    )
    user: Mapped[User] = relationship(
        back_populates="business_profiles",
    )
    embedding: Mapped[BusinessProfileEmbedding | None] = relationship(
        back_populates="business_profile",
        cascade="all, delete-orphan",
        passive_deletes=True,
        uselist=False,
    )
    matches: Mapped[list[TenderMatch]] = relationship(
        back_populates="business_profile",
        cascade="all, delete-orphan",
        passive_deletes=True,
    )
    saved_tenders: Mapped[list[SavedTender]] = relationship(
        back_populates="business_profile",
    )
    dismissed_tenders: Mapped[list[DismissedTender]] = relationship(
        back_populates="business_profile",
    )
    processing_errors: Mapped[list[ProcessingError]] = relationship(
        back_populates="business_profile",
    )

		

PostgreSQL’s SQLAlchemy dialect supports both JSON and JSONB, including PostgreSQL-specific JSONB operations.

Why Use JSONB for Profile Lists?

The following profile fields contain flexible lists:

			
services
target_industries
preferred_countries
preferred_regions
included_keywords
excluded_keywords
certifications
preferred_languages

		

For example:

			
[
  "cloud migration",
  "managed IT services",
  "cybersecurity consulting"
]

		

Using JSONB keeps the initial schema manageable.

We do not yet need separate relational tables for every service, country, keyword, or certification.

That may change if we later introduce:

Controlled taxonomies
Advanced faceted search
Reporting across profiles
Shared keyword libraries
Formal certification entities

Important JSONB Update Behavior

When updating a JSONB list, prefer assigning a new list:

			
profile.services = [
    *profile.services,
    "Microsoft 365 consulting",
]

instead of only mutating the existing list:

profile.services.append("Microsoft 365 consulting")

Plain SQLAlchemy JSONB columns may not automatically detect every in-place list mutation unless mutable tracking is configured.

Replacing the complete list avoids ambiguity during the MVP.

Step 6: Create the Tender Source Model

Create:

app/models/tender_source.py

Add:

			
from __future__ import annotations
from typing import TYPE_CHECKING
from sqlalchemy import Boolean, String, Text
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.base import Base
from app.models.mixins import TimestampMixin, UUIDPrimaryKeyMixin
if TYPE_CHECKING:
    from app.models.ingestion import IngestionRun, RawTenderRecord
    from app.models.processing_error import ProcessingError
    from app.models.tender import Tender
class TenderSource(
    UUIDPrimaryKeyMixin,
    TimestampMixin,
    Base,
):
    __tablename__ = "tender_sources"
    name: Mapped[str] = mapped_column(
        String(255),
        nullable=False,
    )
    slug: Mapped[str] = mapped_column(
        String(100),
        nullable=False,
        unique=True,
        index=True,
    )
    source_type: Mapped[str] = mapped_column(
        String(50),
        nullable=False,
        index=True,
    )
    base_url: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
    )
    country_code: Mapped[str | None] = mapped_column(
        String(2),
        nullable=True,
        index=True,
    )
    is_active: Mapped[bool] = mapped_column(
        Boolean,
        default=True,
        server_default="true",
        nullable=False,
        index=True,
    )
    configuration: Mapped[dict[str, object]] = mapped_column(
        JSONB,
        default=dict,
        nullable=False,
    )
    ingestion_runs: Mapped[list[IngestionRun]] = relationship(
        back_populates="source",
    )
    raw_records: Mapped[list[RawTenderRecord]] = relationship(
        back_populates="source",
    )
    tenders: Mapped[list[Tender]] = relationship(
        back_populates="source",
    )
    processing_errors: Mapped[list[ProcessingError]] = relationship(
        back_populates="source",
    )

		

Source Configuration and Secrets

The configuration field may contain non-sensitive source settings:

			
{
  "page_size": 100,
  "default_language": "en",
  "supports_incremental_updates": true
}

		

It should not contain:

API passwords
Authentication tokens
Private keys
Database credentials

Secrets belong in environment variables or a production secret-management service.

Step 7: Create the Ingestion Models

Create:

app/models/ingestion.py

Add:

			
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING
from uuid import UUID
from sqlalchemy import (
    CheckConstraint,
    DateTime,
    ForeignKey,
    Index,
    Integer,
    String,
    Text,
    UniqueConstraint,
    func,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.base import Base
from app.models.mixins import TimestampMixin, UUIDPrimaryKeyMixin
if TYPE_CHECKING:
    from app.models.processing_error import ProcessingError
    from app.models.tender import Tender
    from app.models.tender_source import TenderSource
class IngestionRun(
    UUIDPrimaryKeyMixin,
    TimestampMixin,
    Base,
):
    __tablename__ = "ingestion_runs"
    __table_args__ = (
        CheckConstraint(
            "records_retrieved >= 0",
            name="records_retrieved_non_negative",
        ),
        CheckConstraint(
            "records_created >= 0",
            name="records_created_non_negative",
        ),
        CheckConstraint(
            "records_updated >= 0",
            name="records_updated_non_negative",
        ),
        CheckConstraint(
            "records_skipped >= 0",
            name="records_skipped_non_negative",
        ),
        CheckConstraint(
            "records_failed >= 0",
            name="records_failed_non_negative",
        ),
        Index(
            "ix_ingestion_runs_source_status_started",
            "source_id",
            "status",
            "started_at",
        ),
    )
    source_id: Mapped[UUID] = mapped_column(
        ForeignKey(
            "tender_sources.id",
            ondelete="RESTRICT",
        ),
        nullable=False,
        index=True,
    )
    status: Mapped[str] = mapped_column(
        String(40),
        nullable=False,
        index=True,
    )
    trigger_type: Mapped[str] = mapped_column(
        String(40),
        nullable=False,
    )
    started_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )
    completed_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
        nullable=True,
    )
    records_retrieved: Mapped[int] = mapped_column(
        Integer,
        default=0,
        server_default="0",
        nullable=False,
    )
    records_created: Mapped[int] = mapped_column(
        Integer,
        default=0,
        server_default="0",
        nullable=False,
    )
    records_updated: Mapped[int] = mapped_column(
        Integer,
        default=0,
        server_default="0",
        nullable=False,
    )
    records_skipped: Mapped[int] = mapped_column(
        Integer,
        default=0,
        server_default="0",
        nullable=False,
    )
    records_failed: Mapped[int] = mapped_column(
        Integer,
        default=0,
        server_default="0",
        nullable=False,
    )
    cursor_value: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
    )
    error_summary: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
    )
    source: Mapped[TenderSource] = relationship(
        back_populates="ingestion_runs",
    )
    raw_records: Mapped[list[RawTenderRecord]] = relationship(
        back_populates="ingestion_run",
    )
    processing_errors: Mapped[list[ProcessingError]] = relationship(
        back_populates="ingestion_run",
    )
class RawTenderRecord(
    UUIDPrimaryKeyMixin,
    TimestampMixin,
    Base,
):
    __tablename__ = "raw_tender_records"
    __table_args__ = (
        UniqueConstraint(
            "source_id",
            "external_id",
            name="source_external_id",
        ),
        Index(
            "ix_raw_tender_records_processing",
            "processing_status",
            "retrieved_at",
        ),
    )
    source_id: Mapped[UUID] = mapped_column(
        ForeignKey(
            "tender_sources.id",
            ondelete="RESTRICT",
        ),
        nullable=False,
        index=True,
    )
    ingestion_run_id: Mapped[UUID | None] = mapped_column(
        ForeignKey(
            "ingestion_runs.id",
            ondelete="SET NULL",
        ),
        nullable=True,
        index=True,
    )
    external_id: Mapped[str] = mapped_column(
        String(255),
        nullable=False,
    )
    source_url: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
    )
    payload: Mapped[dict[str, object]] = mapped_column(
        JSONB,
        nullable=False,
    )
    payload_hash: Mapped[str | None] = mapped_column(
        String(64),
        nullable=True,
        index=True,
    )
    retrieved_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )
    processing_status: Mapped[str] = mapped_column(
        String(40),
        default="retrieved",
        server_default="retrieved",
        nullable=False,
        index=True,
    )
    processing_attempts: Mapped[int] = mapped_column(
        Integer,
        default=0,
        server_default="0",
        nullable=False,
    )
    last_error: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
    )
    source: Mapped[TenderSource] = relationship(
        back_populates="raw_records",
    )
    ingestion_run: Mapped[IngestionRun | None] = relationship(
        back_populates="raw_records",
    )
    tender: Mapped[Tender | None] = relationship(
        back_populates="raw_record",
        uselist=False,
    )
    processing_errors: Mapped[list[ProcessingError]] = relationship(
        back_populates="raw_tender_record",
    )

		

Why Preserve Raw Tender Records?

The raw record stores the source payload exactly as received.

The normalized tender stores the fields the application understands.

These layers solve different problems.

The raw record supports:

Debugging
Auditing
Reprocessing
Source-format changes
Extraction of new fields
Detection of updated notices

The normalized tender supports:

Dashboard queries
Matching
Filtering
AI analysis
API responses

Step 8: Create the Tender Model

Create:

app/models/tender.py

Add:

			
from __future__ import annotations
from datetime import date, datetime
from decimal import Decimal
from typing import TYPE_CHECKING
from uuid import UUID
from sqlalchemy import (
    CheckConstraint,
    Date,
    DateTime,
    ForeignKey,
    Index,
    Numeric,
    String,
    Text,
    UniqueConstraint,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.base import Base
from app.models.mixins import TimestampMixin, UUIDPrimaryKeyMixin
if TYPE_CHECKING:
    from app.models.embedding import TenderEmbedding
    from app.models.ingestion import RawTenderRecord
    from app.models.processing_error import ProcessingError
    from app.models.tender_match import TenderMatch
    from app.models.tender_source import TenderSource
    from app.models.tender_summary import TenderSummary
    from app.models.user_tender_action import (
        DismissedTender,
        SavedTender,
    )
class Tender(
    UUIDPrimaryKeyMixin,
    TimestampMixin,
    Base,
):
    __tablename__ = "tenders"
    __table_args__ = (
        UniqueConstraint(
            "source_id",
            "external_id",
            name="source_external_id",
        ),
        CheckConstraint(
            "estimated_value IS NULL OR estimated_value >= 0",
            name="estimated_value_non_negative",
        ),
        Index(
            "ix_tenders_status_deadline",
            "notice_status",
            "submission_deadline",
        ),
        Index(
            "ix_tenders_country_deadline",
            "country_code",
            "submission_deadline",
        ),
        Index(
            "ix_tenders_processing_status_updated",
            "processing_status",
            "updated_at",
        ),
    )
    source_id: Mapped[UUID] = mapped_column(
        ForeignKey(
            "tender_sources.id",
            ondelete="RESTRICT",
        ),
        nullable=False,
        index=True,
    )
    raw_record_id: Mapped[UUID | None] = mapped_column(
        ForeignKey(
            "raw_tender_records.id",
            ondelete="SET NULL",
        ),
        nullable=True,
        unique=True,
    )
    external_id: Mapped[str] = mapped_column(
        String(255),
        nullable=False,
    )
    reference_number: Mapped[str | None] = mapped_column(
        String(255),
        nullable=True,
        index=True,
    )
    title: Mapped[str] = mapped_column(
        Text,
        nullable=False,
    )
    description: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
    )
    contracting_authority: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
        index=True,
    )
    buyer_type: Mapped[str | None] = mapped_column(
        String(100),
        nullable=True,
    )
    country_code: Mapped[str | None] = mapped_column(
        String(2),
        nullable=True,
        index=True,
    )
    region: Mapped[str | None] = mapped_column(
        String(255),
        nullable=True,
        index=True,
    )
    city: Mapped[str | None] = mapped_column(
        String(255),
        nullable=True,
    )
    publication_date: Mapped[date | None] = mapped_column(
        Date,
        nullable=True,
        index=True,
    )
    submission_deadline: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
        nullable=True,
        index=True,
    )
    contract_start_date: Mapped[date | None] = mapped_column(
        Date,
        nullable=True,
    )
    contract_end_date: Mapped[date | None] = mapped_column(
        Date,
        nullable=True,
    )
    estimated_value: Mapped[Decimal | None] = mapped_column(
        Numeric(18, 2),
        nullable=True,
        index=True,
    )
    currency: Mapped[str | None] = mapped_column(
        String(3),
        nullable=True,
    )
    procurement_category: Mapped[str | None] = mapped_column(
        String(255),
        nullable=True,
        index=True,
    )
    cpv_codes: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    language_code: Mapped[str | None] = mapped_column(
        String(10),
        nullable=True,
    )
    source_url: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
    )
    notice_status: Mapped[str] = mapped_column(
        String(40),
        default="active",
        server_default="active",
        nullable=False,
        index=True,
    )
    processing_status: Mapped[str] = mapped_column(
        String(40),
        default="analysis_pending",
        server_default="analysis_pending",
        nullable=False,
        index=True,
    )
    source: Mapped[TenderSource] = relationship(
        back_populates="tenders",
    )
    raw_record: Mapped[RawTenderRecord | None] = relationship(
        back_populates="tender",
    )
    summary: Mapped[TenderSummary | None] = relationship(
        back_populates="tender",
        cascade="all, delete-orphan",
        passive_deletes=True,
        uselist=False,
    )
    embedding: Mapped[TenderEmbedding | None] = relationship(
        back_populates="tender",
        cascade="all, delete-orphan",
        passive_deletes=True,
        uselist=False,
    )
    matches: Mapped[list[TenderMatch]] = relationship(
        back_populates="tender",
        cascade="all, delete-orphan",
        passive_deletes=True,
    )
    saved_by_users: Mapped[list[SavedTender]] = relationship(
        back_populates="tender",
    )
    dismissed_by_users: Mapped[list[DismissedTender]] = relationship(
        back_populates="tender",
    )
    processing_errors: Mapped[list[ProcessingError]] = relationship(
        back_populates="tender",
    )

		

Why Use `Numeric` for Contract Values?

Financial values should not use floating-point columns.

Floating-point representations may introduce rounding artifacts.

We therefore use:

Numeric(18, 2)

and represent the corresponding Python value as:

Decimal

This provides predictable precision for contract values.

Why Are Many Tender Fields Nullable?

Public procurement sources do not always provide the same fields.

A notice may omit:

Estimated value
Region
Contract dates
Buyer type
Currency
Category
Submission deadline

The database should represent missing information honestly.

Using a false default such as zero for a missing contract value would distort filtering and match scoring.

Step 9: Create the Tender Summary Model

Create:

app/models/tender_summary.py

Add:

			
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING
from uuid import UUID
from sqlalchemy import DateTime, ForeignKey, Integer, String, Text
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.base import Base
from app.models.mixins import TimestampMixin, UUIDPrimaryKeyMixin
if TYPE_CHECKING:
    from app.models.tender import Tender
class TenderSummary(
    UUIDPrimaryKeyMixin,
    TimestampMixin,
    Base,
):
    __tablename__ = "tender_summaries"
    tender_id: Mapped[UUID] = mapped_column(
        ForeignKey(
            "tenders.id",
            ondelete="CASCADE",
        ),
        nullable=False,
        unique=True,
        index=True,
    )
    overview: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
    )
    deliverables: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    requirements: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    certifications: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    language_requirements: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    risks: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    contract_duration: Mapped[str | None] = mapped_column(
        String(255),
        nullable=True,
    )
    missing_information: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    model_name: Mapped[str] = mapped_column(
        String(150),
        nullable=False,
    )
    prompt_version: Mapped[str] = mapped_column(
        String(50),
        nullable=False,
    )
    input_hash: Mapped[str | None] = mapped_column(
        String(64),
        nullable=True,
        index=True,
    )
    input_tokens: Mapped[int | None] = mapped_column(
        Integer,
        nullable=True,
    )
    output_tokens: Mapped[int | None] = mapped_column(
        Integer,
        nullable=True,
    )
    status: Mapped[str] = mapped_column(
        String(40),
        default="pending",
        server_default="pending",
        nullable=False,
        index=True,
    )
    generated_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
        nullable=True,
    )
    tender: Mapped[Tender] = relationship(
        back_populates="summary",
    )

		

Why Keep AI Summaries Separate?

The tender row represents normalized procurement data.

The summary row represents derived AI output.

Separating them allows us to:

Regenerate a summary
Change the prompt
Change the AI model
Track token usage
Detect stale output
Retry failed analysis
Keep source-derived data independent of AI-derived data

The unique constraint on tender_id means the MVP stores one current summary per tender.

A later version can introduce a summary-history table if complete version retention becomes necessary.

Step 10: Define the Embedding Dimension

The PostgreSQL vector column requires a fixed number of dimensions.

Create:

app/core/constants.py

Add:

EMBEDDING_DIMENSIONS = 1536

This value must match the embedding model selected for the application.

Do not change the embedding model without also considering:

The database vector dimension
Existing tender vectors
Existing business-profile vectors
Vector indexes
Match calculations
Re-embedding requirements

Different vector dimensions generally require a schema migration or a separate embedding table.

Step 11: Create the Embedding Models

Create:

app/models/embedding.py

Add:

			
from __future__ import annotations
from typing import TYPE_CHECKING
from uuid import UUID
from pgvector.sqlalchemy import VECTOR
from sqlalchemy import ForeignKey, Integer, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.constants import EMBEDDING_DIMENSIONS
from app.database.base import Base
from app.models.mixins import TimestampMixin, UUIDPrimaryKeyMixin
if TYPE_CHECKING:
    from app.models.business_profile import BusinessProfile
    from app.models.tender import Tender
class TenderEmbedding(
    UUIDPrimaryKeyMixin,
    TimestampMixin,
    Base,
):
    __tablename__ = "tender_embeddings"
    tender_id: Mapped[UUID] = mapped_column(
        ForeignKey(
            "tenders.id",
            ondelete="CASCADE",
        ),
        nullable=False,
        unique=True,
        index=True,
    )
    embedding: Mapped[list[float]] = mapped_column(
        VECTOR(EMBEDDING_DIMENSIONS),
        nullable=False,
    )
    model_name: Mapped[str] = mapped_column(
        String(150),
        nullable=False,
    )
    embedding_dimensions: Mapped[int] = mapped_column(
        Integer,
        default=EMBEDDING_DIMENSIONS,
        nullable=False,
    )
    content_hash: Mapped[str | None] = mapped_column(
        String(64),
        nullable=True,
        index=True,
    )
    tender: Mapped[Tender] = relationship(
        back_populates="embedding",
    )
class BusinessProfileEmbedding(
    UUIDPrimaryKeyMixin,
    TimestampMixin,
    Base,
):
    __tablename__ = "business_profile_embeddings"
    business_profile_id: Mapped[UUID] = mapped_column(
        ForeignKey(
            "business_profiles.id",
            ondelete="CASCADE",
        ),
        nullable=False,
        unique=True,
        index=True,
    )
    embedding: Mapped[list[float]] = mapped_column(
        VECTOR(EMBEDDING_DIMENSIONS),
        nullable=False,
    )
    model_name: Mapped[str] = mapped_column(
        String(150),
        nullable=False,
    )
    embedding_dimensions: Mapped[int] = mapped_column(
        Integer,
        default=EMBEDDING_DIMENSIONS,
        nullable=False,
    )
    content_hash: Mapped[str | None] = mapped_column(
        String(64),
        nullable=True,
        index=True,
    )
    business_profile: Mapped[BusinessProfile] = relationship(
        back_populates="embedding",
    )

		

The official pgvector Python integration demonstrates importing VECTOR from pgvector.sqlalchemy for SQLAlchemy model columns.

Why Store the Model Name and Content Hash?

The vector itself is not enough to manage embedding lifecycle.

We also store:

			
model_name
embedding_dimensions
content_hash

These fields allow the application to determine whether an embedding is stale.

For example:

			
Current content hash = stored content hash
Current model = stored model

No regeneration is necessary.

However:

Current content hash != stored content hash

The source text changed and should be embedded again.

Similarly:

Current model != stored model

The system may need to regenerate the vector using the selected model.

Step 12: Create the Tender Match Model

Create:

app/models/tender_match.py

Add:

			
from __future__ import annotations
from datetime import datetime
from decimal import Decimal
from typing import TYPE_CHECKING
from uuid import UUID
from sqlalchemy import (
    CheckConstraint,
    DateTime,
    ForeignKey,
    Index,
    Numeric,
    String,
    UniqueConstraint,
    func,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.base import Base
from app.models.mixins import TimestampMixin, UUIDPrimaryKeyMixin
if TYPE_CHECKING:
    from app.models.business_profile import BusinessProfile
    from app.models.tender import Tender
class TenderMatch(
    UUIDPrimaryKeyMixin,
    TimestampMixin,
    Base,
):
    __tablename__ = "tender_matches"
    __table_args__ = (
        UniqueConstraint(
            "business_profile_id",
            "tender_id",
            name="profile_tender",
        ),
        CheckConstraint(
            "overall_score >= 0 AND overall_score <= 100",
            name="overall_score_range",
        ),
        CheckConstraint(
            "semantic_score IS NULL "
            "OR semantic_score BETWEEN 0 AND 100",
            name="semantic_score_range",
        ),
        CheckConstraint(
            "keyword_score IS NULL "
            "OR keyword_score BETWEEN 0 AND 100",
            name="keyword_score_range",
        ),
        CheckConstraint(
            "service_score IS NULL "
            "OR service_score BETWEEN 0 AND 100",
            name="service_score_range",
        ),
        CheckConstraint(
            "industry_score IS NULL "
            "OR industry_score BETWEEN 0 AND 100",
            name="industry_score_range",
        ),
        CheckConstraint(
            "geography_score IS NULL "
            "OR geography_score BETWEEN 0 AND 100",
            name="geography_score_range",
        ),
        CheckConstraint(
            "value_score IS NULL "
            "OR value_score BETWEEN 0 AND 100",
            name="value_score_range",
        ),
        CheckConstraint(
            "certification_score IS NULL "
            "OR certification_score BETWEEN 0 AND 100",
            name="certification_score_range",
        ),
        CheckConstraint(
            "deadline_score IS NULL "
            "OR deadline_score BETWEEN 0 AND 100",
            name="deadline_score_range",
        ),
        CheckConstraint(
            "exclusion_penalty >= 0",
            name="exclusion_penalty_non_negative",
        ),
        Index(
            "ix_tender_matches_profile_score",
            "business_profile_id",
            "overall_score",
        ),
        Index(
            "ix_tender_matches_profile_status",
            "business_profile_id",
            "status",
        ),
    )
    business_profile_id: Mapped[UUID] = mapped_column(
        ForeignKey(
            "business_profiles.id",
            ondelete="CASCADE",
        ),
        nullable=False,
        index=True,
    )
    tender_id: Mapped[UUID] = mapped_column(
        ForeignKey(
            "tenders.id",
            ondelete="CASCADE",
        ),
        nullable=False,
        index=True,
    )
    overall_score: Mapped[Decimal] = mapped_column(
        Numeric(5, 2),
        nullable=False,
        index=True,
    )
    semantic_score: Mapped[Decimal | None] = mapped_column(
        Numeric(5, 2),
        nullable=True,
    )
    keyword_score: Mapped[Decimal | None] = mapped_column(
        Numeric(5, 2),
        nullable=True,
    )
    service_score: Mapped[Decimal | None] = mapped_column(
        Numeric(5, 2),
        nullable=True,
    )
    industry_score: Mapped[Decimal | None] = mapped_column(
        Numeric(5, 2),
        nullable=True,
    )
    geography_score: Mapped[Decimal | None] = mapped_column(
        Numeric(5, 2),
        nullable=True,
    )
    value_score: Mapped[Decimal | None] = mapped_column(
        Numeric(5, 2),
        nullable=True,
    )
    certification_score: Mapped[Decimal | None] = mapped_column(
        Numeric(5, 2),
        nullable=True,
    )
    deadline_score: Mapped[Decimal | None] = mapped_column(
        Numeric(5, 2),
        nullable=True,
    )
    exclusion_penalty: Mapped[Decimal] = mapped_column(
        Numeric(5, 2),
        default=0,
        server_default="0",
        nullable=False,
    )
    strong_matches: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    potential_gaps: Mapped[list[str]] = mapped_column(
        JSONB,
        default=list,
        nullable=False,
    )
    score_version: Mapped[str] = mapped_column(
        String(50),
        nullable=False,
    )
    status: Mapped[str] = mapped_column(
        String(40),
        default="active",
        server_default="active",
        nullable=False,
        index=True,
    )
    calculated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )
    business_profile: Mapped[BusinessProfile] = relationship(
        back_populates="matches",
    )
    tender: Mapped[Tender] = relationship(
        back_populates="matches",
    )

		

SQLAlchemy supports table-level constraints and indexes through __table_args__, which we use to enforce score ranges and uniqueness.

Why Store Individual Match Components?

The user will eventually see a single overall match score.

However, the application needs more detail internally.

For example:

			
Overall score:          82
Semantic score:         91
Service score:          88
Geography score:       100
Contract value score:   70
Certification score:    40
Exclusion penalty:       5

		

This allows us to explain:

Why a tender ranked highly
Which requirements matched
Which requirements created gaps
Whether an exclusion rule reduced the score
Whether the semantic model overestimated relevance
Which scoring version produced the result

Explainability is especially important because the system combines deterministic rules with AI-generated semantic similarity.

Step 13: Create Saved and Dismissed Tender Models

Create:

app/models/user_tender_action.py

Add:

			
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING
from uuid import UUID
from sqlalchemy import (
    DateTime,
    ForeignKey,
    String,
    Text,
    UniqueConstraint,
    func,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.base import Base
from app.models.mixins import TimestampMixin, UUIDPrimaryKeyMixin
if TYPE_CHECKING:
    from app.models.business_profile import BusinessProfile
    from app.models.tender import Tender
    from app.models.user import User
class SavedTender(
    UUIDPrimaryKeyMixin,
    TimestampMixin,
    Base,
):
    __tablename__ = "saved_tenders"
    __table_args__ = (
        UniqueConstraint(
            "user_id",
            "tender_id",
            name="user_tender",
        ),
    )
    user_id: Mapped[UUID] = mapped_column(
        ForeignKey(
            "users.id",
            ondelete="CASCADE",
        ),
        nullable=False,
        index=True,
    )
    tender_id: Mapped[UUID] = mapped_column(
        ForeignKey(
            "tenders.id",
            ondelete="CASCADE",
        ),
        nullable=False,
        index=True,
    )
    business_profile_id: Mapped[UUID | None] = mapped_column(
        ForeignKey(
            "business_profiles.id",
            ondelete="SET NULL",
        ),
        nullable=True,
        index=True,
    )
    note: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
    )
    saved_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
        index=True,
    )
    user: Mapped[User] = relationship(
        back_populates="saved_tenders",
    )
    tender: Mapped[Tender] = relationship(
        back_populates="saved_by_users",
    )
    business_profile: Mapped[BusinessProfile | None] = relationship(
        back_populates="saved_tenders",
    )
class DismissedTender(
    UUIDPrimaryKeyMixin,
    TimestampMixin,
    Base,
):
    __tablename__ = "dismissed_tenders"
    __table_args__ = (
        UniqueConstraint(
            "user_id",
            "tender_id",
            name="user_tender",
        ),
    )
    user_id: Mapped[UUID] = mapped_column(
        ForeignKey(
            "users.id",
            ondelete="CASCADE",
        ),
        nullable=False,
        index=True,
    )
    tender_id: Mapped[UUID] = mapped_column(
        ForeignKey(
            "tenders.id",
            ondelete="CASCADE",
        ),
        nullable=False,
        index=True,
    )
    business_profile_id: Mapped[UUID | None] = mapped_column(
        ForeignKey(
            "business_profiles.id",
            ondelete="SET NULL",
        ),
        nullable=True,
        index=True,
    )
    reason_code: Mapped[str | None] = mapped_column(
        String(50),
        nullable=True,
        index=True,
    )
    reason_text: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
    )
    dismissed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
        index=True,
    )
    user: Mapped[User] = relationship(
        back_populates="dismissed_tenders",
    )
    tender: Mapped[Tender] = relationship(
        back_populates="dismissed_by_users",
    )
    business_profile: Mapped[BusinessProfile | None] = relationship(
        back_populates="dismissed_tenders",
    )

		

Why Keep Saved and Dismissed Tenders Separate?

We could store a single status on tender_matches.

However, saving and dismissing are explicit user actions.

Separate tables allow us to store:

The user who performed the action
The business profile context
The action timestamp
A private note
A dismissal reason
Future workflow metadata

It also keeps matching results separate from user workflow state.

A tender may be rematched many times without losing the fact that the user previously saved or dismissed it.

Step 14: Create the Processing Error Model

Create:

app/models/processing_error.py

Add:

			
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING
from uuid import UUID
from sqlalchemy import (
    Boolean,
    DateTime,
    ForeignKey,
    Index,
    Integer,
    String,
    Text,
    func,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database.base import Base
from app.models.mixins import TimestampMixin, UUIDPrimaryKeyMixin
if TYPE_CHECKING:
    from app.models.business_profile import BusinessProfile
    from app.models.ingestion import IngestionRun, RawTenderRecord
    from app.models.tender import Tender
    from app.models.tender_source import TenderSource
class ProcessingError(
    UUIDPrimaryKeyMixin,
    TimestampMixin,
    Base,
):
    __tablename__ = "processing_errors"
    __table_args__ = (
        Index(
            "ix_processing_errors_unresolved_stage",
            "resolved",
            "stage",
            "occurred_at",
        ),
    )
    stage: Mapped[str] = mapped_column(
        String(50),
        nullable=False,
        index=True,
    )
    source_id: Mapped[UUID | None] = mapped_column(
        ForeignKey(
            "tender_sources.id",
            ondelete="SET NULL",
        ),
        nullable=True,
        index=True,
    )
    ingestion_run_id: Mapped[UUID | None] = mapped_column(
        ForeignKey(
            "ingestion_runs.id",
            ondelete="SET NULL",
        ),
        nullable=True,
        index=True,
    )
    raw_tender_record_id: Mapped[UUID | None] = mapped_column(
        ForeignKey(
            "raw_tender_records.id",
            ondelete="SET NULL",
        ),
        nullable=True,
        index=True,
    )
    tender_id: Mapped[UUID | None] = mapped_column(
        ForeignKey(
            "tenders.id",
            ondelete="SET NULL",
        ),
        nullable=True,
        index=True,
    )
    business_profile_id: Mapped[UUID | None] = mapped_column(
        ForeignKey(
            "business_profiles.id",
            ondelete="SET NULL",
        ),
        nullable=True,
        index=True,
    )
    error_type: Mapped[str] = mapped_column(
        String(255),
        nullable=False,
        index=True,
    )
    error_message: Mapped[str] = mapped_column(
        Text,
        nullable=False,
    )
    error_details: Mapped[dict[str, object]] = mapped_column(
        JSONB,
        default=dict,
        nullable=False,
    )
    retryable: Mapped[bool] = mapped_column(
        Boolean,
        default=False,
        server_default="false",
        nullable=False,
    )
    retry_count: Mapped[int] = mapped_column(
        Integer,
        default=0,
        server_default="0",
        nullable=False,
    )
    resolved: Mapped[bool] = mapped_column(
        Boolean,
        default=False,
        server_default="false",
        nullable=False,
        index=True,
    )
    occurred_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )
    resolved_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
        nullable=True,
    )
    source: Mapped[TenderSource | None] = relationship(
        back_populates="processing_errors",
    )
    ingestion_run: Mapped[IngestionRun | None] = relationship(
        back_populates="processing_errors",
    )
    raw_tender_record: Mapped[RawTenderRecord | None] = relationship(
        back_populates="processing_errors",
    )
    tender: Mapped[Tender | None] = relationship(
        back_populates="processing_errors",
    )
    business_profile: Mapped[BusinessProfile | None] = relationship(
        back_populates="processing_errors",
    )

		

What Should Be Stored in `error_details`?

The error_details JSONB field may contain structured diagnostic information:

			
{
  "http_status": 429,
  "provider": "ai-provider",
  "operation": "generate_tender_summary",
  "attempt": 3
}

		

Do not store secrets or complete sensitive request headers in error details.

The field should help diagnose the problem without exposing credentials.

Step 15: Import Every Model

Alembic can only detect tables that have been imported into SQLAlchemy’s metadata.

Open:

app/models/__init__.py

Add:

			
from app.models.business_profile import BusinessProfile
from app.models.embedding import (
    BusinessProfileEmbedding,
    TenderEmbedding,
)
from app.models.ingestion import IngestionRun, RawTenderRecord
from app.models.processing_error import ProcessingError
from app.models.tender import Tender
from app.models.tender_match import TenderMatch
from app.models.tender_source import TenderSource
from app.models.tender_summary import TenderSummary
from app.models.user import User
from app.models.user_tender_action import (
    DismissedTender,
    SavedTender,
)
__all__ = [
    "BusinessProfile",
    "BusinessProfileEmbedding",
    "DismissedTender",
    "IngestionRun",
    "ProcessingError",
    "RawTenderRecord",
    "SavedTender",
    "Tender",
    "TenderEmbedding",
    "TenderMatch",
    "TenderSource",
    "TenderSummary",
    "User",
]

		

This import file ensures that all model classes are registered with:

Base.metadata

Alembic compares that metadata against the current database when generating candidate migrations.

Step 16: Create a Metadata Loader

Open:

app/database/base.py

Add a model import function below the base class:

			
from sqlalchemy import MetaData
from sqlalchemy.orm import DeclarativeBase
NAMING_CONVENTION = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": (
        "fk_%(table_name)s_%(column_0_name)s_"
        "%(referred_table_name)s"
    ),
    "pk": "pk_%(table_name)s",
}
class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=NAMING_CONVENTION)
def import_models() -> None:
    import app.models  # noqa: F401

		

We will call this function from the Alembic configuration in the next article.

Step 17: Verify Model Registration

Create:

scripts/check_models.py

Add:

			
from app.database.base import Base, import_models
def main() -> None:
    import_models()
    table_names = sorted(Base.metadata.tables.keys())
    print("Registered SQLAlchemy tables:")
    for table_name in table_names:
        print(f"- {table_name}")
    print(f"\nTotal tables: {len(table_names)}")
if __name__ == "__main__":
    main()

		

Run it from the backend directory:

python ../scripts/check_models.py

Depending on your repository layout, you can also place the script inside:

backend/scripts/check_models.py

and run:

python scripts/check_models.py

Expected Registered Tables

The output should include:

			
Registered SQLAlchemy tables:
- business_profile_embeddings
- business_profiles
- dismissed_tenders
- ingestion_runs
- processing_errors
- raw_tender_records
- saved_tenders
- tender_embeddings
- tender_matches
- tender_sources
- tender_summaries
- tenders
- users
Total tables: 13

		

If a table is missing, confirm that its model is imported in:

app/models/__init__.py

Step 18: Add a Model Metadata Test

Create:

tests/models/test_metadata.py

Add:

			
from app.database.base import Base, import_models
EXPECTED_TABLES = {
    "users",
    "business_profiles",
    "tender_sources",
    "ingestion_runs",
    "raw_tender_records",
    "tenders",
    "tender_summaries",
    "tender_embeddings",
    "business_profile_embeddings",
    "tender_matches",
    "saved_tenders",
    "dismissed_tenders",
    "processing_errors",
}
def test_all_expected_tables_are_registered() -> None:
    import_models()
    registered_tables = set(Base.metadata.tables.keys())
    assert registered_tables == EXPECTED_TABLES

		

Create the models test directory if it does not exist:

mkdir tests/models

Also create:

tests/models/__init__.py

Run:

pytest

The health-check test and metadata test should both pass.

Current Relationship Overview

The implemented SQLAlchemy relationships now represent the following structure:

			
User
 |
 +-- BusinessProfile
 |      |
 |      +-- BusinessProfileEmbedding
 |      |
 |      +-- TenderMatch
 |      |
 |      +-- SavedTender
 |      |
 |      +-- DismissedTender
 |
 +-- SavedTender
 |
 +-- DismissedTender
TenderSource
 |
 +-- IngestionRun
 |
 +-- RawTenderRecord
 |
 +-- Tender
RawTenderRecord
 |
 +-- Tender
Tender
 |
 +-- TenderSummary
 |
 +-- TenderEmbedding
 |
 +-- TenderMatch
 |
 +-- SavedTender
 |
 +-- DismissedTender
 |
 +-- ProcessingError

		

One-to-One Relationships

The following relationships are one-to-one:

			
RawTenderRecord -> Tender
Tender -> TenderSummary
Tender -> TenderEmbedding
BusinessProfile -> BusinessProfileEmbedding

These relationships use:

uselist=False

and a unique foreign key.

For example:

			
tender_id = mapped_column(
    ForeignKey("tenders.id"),
    unique=True,
)

The unique database constraint is what truly enforces the one-to-one rule.

Many-to-Many Relationships with Additional Data

Business profiles and tenders have a many-to-many relationship.

However, the join contains important attributes:

Overall score
Semantic score
Geography score
Match explanations
Score version
Calculation time

We therefore model the join as a full entity:

TenderMatch

rather than using a simple association table.

This pattern is sometimes called an association-object relationship.

Deletion Rules

The models use several foreign-key deletion strategies.

`CASCADE`

Used when the child record has no meaning without the parent.

Examples:

			
User deleted -> BusinessProfile deleted
Tender deleted -> TenderSummary deleted
Tender deleted -> TenderEmbedding deleted
Profile deleted -> TenderMatch deleted

`SET NULL`

Used when historical records should remain but their original reference may disappear.

Examples:

			
Ingestion run deleted -> Raw record remains
Profile deleted -> Saved action may remain
Tender deleted -> Processing error may remain

`RESTRICT`

Used for shared system entities that should not be removed while dependent records exist.

Example:

TenderSource

A tender source should generally be deactivated instead of deleted.

Avoiding Accidental Tender Deletion

In production, tenders should rarely be physically deleted.

Instead, change:

notice_status

to a value such as:

			
expired
cancelled
withdrawn
awarded

Physical deletion could remove:

Historical matching evidence
Saved-user workflows
AI-analysis history
Operational error records
Ingestion traceability

The database supports cascades, but application services should use them carefully.

Model Status Values

The current models use string fields for statuses.

Possible values include:

Tender notice status

			
active
expired
awarded
cancelled
withdrawn
unknown

		

Tender processing status

			
analysis_pending
analysis_complete
embedding_pending
embedding_complete
matching_pending
processed
failed

		

Ingestion status

			
pending
running
completed
completed_with_errors
failed
cancelled

		

Raw record status

			
retrieved
normalization_pending
normalized
failed
ignored

		

Summary status

			
pending
processing
completed
failed

Match status

			
active
below_threshold
expired
failed

We will define application-level string enumerations later to prevent typing mistakes.

Why Not Use Native PostgreSQL Enums Yet?

PostgreSQL enums provide strong database validation.

However, they can make migrations more complicated when values change.

For the MVP, we will use:

String columns
Pydantic validation
Application constants
Selected database constraints

When the workflow stabilizes, we can decide whether native database enums provide enough value to justify the additional migration complexity.

Indexing Decisions

The models include indexes for frequent query paths.

Dashboard Queries

			
tender_matches.business_profile_id
tender_matches.overall_score
tenders.notice_status
tenders.submission_deadline

Ingestion Queries

			
raw_tender_records.processing_status
raw_tender_records.retrieved_at
ingestion_runs.source_id
ingestion_runs.status

Filtering

			
tenders.country_code
tenders.region
tenders.estimated_value
tenders.procurement_category

User Actions

			
saved_tenders.user_id
dismissed_tenders.user_id
dismissed_tenders.reason_code

Error Administration

			
processing_errors.resolved
processing_errors.stage
processing_errors.occurred_at

Indexes improve read performance but increase storage and write overhead.

We should measure real query behavior before adding large numbers of additional indexes.

Vector Indexes Are Not Added Yet

The embedding tables currently define vector columns but no approximate-nearest-neighbor indexes.

That is intentional.

pgvector supports exact and approximate similarity search, including several distance measures and index strategies.

Before adding a vector index, we need to know:

The final embedding dimension
The selected distance metric
The expected number of tenders
The required recall
Query latency targets
Available PostgreSQL memory
How frequently vectors are inserted

For an early dataset, exact similarity calculations may be sufficient.

We will add vector indexes only after measuring performance.

Important Schema Design Improvement

The previous conceptual schema included both:

			
source_id
external_id

on raw tender records and normalized tenders.

That remains useful because it provides duplicate protection at both stages.

However, the normalized tender also has:

raw_record_id UNIQUE

This means one raw source record maps to at most one current normalized tender.

If a future procurement source publishes multiple lots inside one notice, we may need to change this rule.

A later design could support:

One raw notice -> multiple normalized lots

For the current MVP, one raw record to one normalized tender is sufficient.

Full Current Model Directory

The backend should now contain:

			
backend/
├── app/
│   ├── core/
│   │   ├── config.py
│   │   ├── constants.py
│   │   ├── exceptions.py
│   │   └── logging.py
│   ├── database/
│   │   ├── base.py
│   │   └── session.py
│   ├── models/
│   │   ├── __init__.py
│   │   ├── business_profile.py
│   │   ├── embedding.py
│   │   ├── ingestion.py
│   │   ├── mixins.py
│   │   ├── processing_error.py
│   │   ├── tender.py
│   │   ├── tender_match.py
│   │   ├── tender_source.py
│   │   ├── tender_summary.py
│   │   ├── user.py
│   │   └── user_tender_action.py
│   └── main.py
├── tests/
│   ├── api/
│   │   └── test_health.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── test_metadata.py
│   └── conftest.py
├── .env
├── .env.example
├── pyproject.toml
└── requirements.txt

		

Running the Tests

Run:

pytest

Expected result:

2 passed

The exact number may be higher if you have added other tests.

At this stage, the tests confirm that:

The FastAPI application imports correctly
The health endpoint responds
Every expected SQLAlchemy table is registered
The model modules do not contain unresolved imports

They do not yet confirm that PostgreSQL can create the schema.

That validation will happen through Alembic and database integration tests.

Common Error: Circular Imports

SQLAlchemy models frequently reference one another.

For example:

			
User -> BusinessProfile
BusinessProfile -> User

Direct imports at runtime can create circular-import errors.

The models use:

from __future__ import annotations

and:

if TYPE_CHECKING:

This allows static type checkers to understand the related types without forcing every related model to be imported immediately during module execution.

Relationship annotations can then use forward references resolved by SQLAlchemy’s declarative mapping system.

Common Error: Model Table Is Missing

If Alembic or the metadata test cannot see a table, check:

app/models/__init__.py

Every model module must be imported before Alembic inspects:

Base.metadata

Defining a class in a file is not enough if that file is never imported.

Common Error: Vector Type Does Not Exist

An error such as:

type "vector" does not exist

means the PostgreSQL pgvector extension has not been enabled.

The database requires:

CREATE EXTENSION IF NOT EXISTS vector;

We will place this command in the initial Alembic migration before creating vector columns.

Common Error: Invalid Embedding Dimension

If the application tries to store a vector with the wrong length, pgvector will reject it.

For example:

			
Database column: VECTOR(1536)
Incoming vector: 3072 values

The embedding model and database dimension must match.

Validate vector length before inserting:

			
if len(embedding) != EMBEDDING_DIMENSIONS:
    raise ValueError(
        "Embedding has an unexpected number of dimensions."
    )

Common Error: Duplicate Tender

The database intentionally rejects duplicate combinations of:

source_id + external_id

When an ingestion process encounters an existing tender, it should:

Load the current record.
Compare the payload or content hash.
Update the record if the source changed.
Skip reprocessing if nothing changed.

It should not blindly insert another row.

Common Error: Check Constraint Failure

A tender match with:

overall_score = 145

will fail because the valid range is zero through one hundred.

Similarly, this profile is invalid:

			
minimum_contract_value = 100000
maximum_contract_value = 50000

These database errors indicate that application validation or scoring logic allowed an invalid value to reach the persistence layer.

Model Layer Boundaries

SQLAlchemy models should describe persistence.

They should not contain:

HTTP response logic
FastAPI dependencies
AI prompts
Source-fetching code
Large matching algorithms
Email delivery logic
Complex workflow orchestration

Those responsibilities belong in:

			
API routes
Services
Repositories
Integrations
Matching modules
Tasks

		

A model may eventually include small domain-friendly properties, but it should not become a container for unrelated application behavior.

What We Have Implemented

The backend now has models for all 13 MVP tables:

			
users
business_profiles
tender_sources
ingestion_runs
raw_tender_records
tenders
tender_summaries
tender_embeddings
business_profile_embeddings
tender_matches
saved_tenders
dismissed_tenders
processing_errors

		

The models include:

Typed SQLAlchemy 2.x mappings
UUID primary keys
Timezone-aware timestamps
PostgreSQL JSONB fields
pgvector columns
Foreign-key relationships
One-to-one mappings
One-to-many mappings
Unique constraints
Check constraints
Query-oriented indexes
Cascade and nullification rules

What We Have Not Implemented Yet

The model files do not create any database tables by themselves.

We still need to:

Initialize Alembic
Connect Alembic to the application settings
Import all model metadata
Enable the pgvector extension
Generate the initial migration
Review the generated migration
Apply the migration to PostgreSQL
Inspect the resulting tables
Test upgrades and downgrades

We also have not yet implemented:

Pydantic request and response schemas
Repositories
CRUD services
Tender source adapters
AI summary generation
Embedding services
Match calculations
API endpoints for domain entities

Implementation Checklist

Before moving forward, confirm:

			
[ ] The pgvector Python package is installed
[ ] The declarative base uses a naming convention
[ ] UUID and timestamp mixins exist
[ ] All 13 SQLAlchemy tables are defined
[ ] All model relationships resolve correctly
[ ] All model modules are imported
[ ] The metadata test passes
[ ] The health test still passes
[ ] The embedding dimension is documented
[ ] No real secrets were added to source control

		

Recommended Git Commit

Create a new development checkpoint:

			
git add .
git commit -m "Add SQLAlchemy models for tender scanner"

This commit should contain only the model-layer work.

The Alembic migration will be created in the next stage as a separate commit.

What We Will Build Next

The SQLAlchemy models now describe the complete MVP database schema.

The next step is connecting these models to Alembic.

We will:

Initialize the Alembic environment
Load the database URL from application settings
Connect Alembic to Base.metadata
Enable the pgvector PostgreSQL extension
Generate the initial schema migration
Inspect the migration before running it
Apply the migration to PostgreSQL
Verify the created tables and constraints
Test migration rollback behavior

The next article in the development series is:

Creating the Initial PostgreSQL Database Migration with Alembic.

Conclusion

The AI Tender Opportunity Scanner now has a complete SQLAlchemy model layer.

The relational structure distinguishes global procurement data from user-owned data.

Tender sources, ingestion runs, and raw records provide traceability from collection through normalization.

Normalized tenders provide a consistent foundation for dashboard queries and AI analysis.

Tender summaries and embeddings remain separate from source-derived tender data, allowing them to be regenerated and versioned independently.

Business profiles describe what each customer is looking for.

Tender matches connect those profiles to global opportunities while preserving score components and explanations.

Saved and dismissed records track explicit user decisions without modifying shared tender data.

Processing errors provide a queryable operational history for ingestion, normalization, AI, embedding, and matching failures.

The model layer now expresses the application’s core data architecture in executable Python.

The next step is turning this metadata into real PostgreSQL tables through Alembic migrations.

Building the SQLAlchemy Models for the AI Tender Opportunity Scanner

Introduction

Development Progress

What We Are Building

Updated Model Directory

Step 1: Install pgvector Support

Step 2: Improve the Declarative Base

Why Add a Naming Convention?

Step 3: Create Shared Model Mixins

Application Defaults and Server Defaults

Why Use UUID Primary Keys?

Step 4: Create the User Model

Understanding the User Relationships

Step 5: Create the Business Profile Model

Why Use JSONB for Profile Lists?

Important JSONB Update Behavior

Step 6: Create the Tender Source Model

Source Configuration and Secrets

Step 7: Create the Ingestion Models

Why Preserve Raw Tender Records?

Step 8: Create the Tender Model

Why Use Numeric for Contract Values?

Why Are Many Tender Fields Nullable?

Step 9: Create the Tender Summary Model

Why Keep AI Summaries Separate?

Step 10: Define the Embedding Dimension

Step 11: Create the Embedding Models

Why Store the Model Name and Content Hash?

Step 12: Create the Tender Match Model

Why Store Individual Match Components?

Step 13: Create Saved and Dismissed Tender Models

Why Keep Saved and Dismissed Tenders Separate?

Step 14: Create the Processing Error Model

What Should Be Stored in error_details?

Step 15: Import Every Model

Step 16: Create a Metadata Loader

Step 17: Verify Model Registration

Expected Registered Tables

Step 18: Add a Model Metadata Test

Current Relationship Overview

One-to-One Relationships

Many-to-Many Relationships with Additional Data

Deletion Rules

CASCADE

SET NULL

RESTRICT

Avoiding Accidental Tender Deletion

Model Status Values

Tender notice status

Tender processing status

Ingestion status

Raw record status

Summary status

Match status

Why Not Use Native PostgreSQL Enums Yet?

Indexing Decisions

Dashboard Queries

Ingestion Queries

Filtering

User Actions

Error Administration

Vector Indexes Are Not Added Yet

Important Schema Design Improvement

Full Current Model Directory

Running the Tests

Common Error: Circular Imports

Common Error: Model Table Is Missing

Common Error: Vector Type Does Not Exist

Common Error: Invalid Embedding Dimension

Common Error: Duplicate Tender

Common Error: Check Constraint Failure

Model Layer Boundaries

What We Have Implemented

What We Have Not Implemented Yet

Implementation Checklist

Recommended Git Commit

What We Will Build Next

Conclusion

Share this:

Like this:

Why Use `Numeric` for Contract Values?

What Should Be Stored in `error_details`?

`CASCADE`

`SET NULL`

`RESTRICT`