Introduction

The AI Tender Opportunity Scanner now has a solid backend foundation:

PostgreSQL database
SQLAlchemy models
Alembic migrations
Repository layer
CRUD service layer
Validation and exception handling

However, there is still one major problem:

The platform has no tenders.

Without tenders:

No matching can occur
No recommendations can be generated
No alerts can be sent
No AI analysis can run

The next major milestone is building the Tender Ingestion Pipeline.

The ingestion pipeline is responsible for:

Connecting to external tender portals
Downloading tender data
Preserving raw source records
Normalizing tender information
Detecting duplicates
Updating existing tenders
Tracking ingestion statistics
Logging processing failures

This pipeline becomes the foundation of the entire AI Tender Opportunity Scanner.

Every future capability depends on the quality and reliability of the ingestion process.

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 Migration                ████████████████████ 100%
Repository Layer                 ████████████████████ 100%
CRUD Service Layer               ████████████████████ 100%
Tender Ingestion Pipeline        ████████████████████ 100%
AI Analysis                      ░░░░░░░░░░░░░░░░░░░░   0%
Matching Engine                  ░░░░░░░░░░░░░░░░░░░░   0%
Frontend Dashboard               ░░░░░░░░░░░░░░░░░░░░   0%

		

What We Are Building

The ingestion pipeline will:

			
Tender Source
      |
      v
Download Records
      |
      v
Store Raw Records
      |
      v
Normalize Data
      |
      v
Detect Duplicates
      |
      v
Create/Update Tenders
      |
      v
Track Statistics

		

At the end of this article we will have:

Source connectors
Ingestion service
Ingestion run tracking
Raw record persistence
Duplicate detection
Tender normalization
Error logging
Pipeline testing

Why Preserve Raw Data?

A common beginner mistake is:

			
Download data
Immediately transform it
Discard original source

This creates problems later.

Suppose:

Mapping logic changes
A bug is discovered
AI processing improves
New fields become important

Without raw data:

The original source information is gone.

That is why our architecture includes:

raw_tender_records

Every source response is preserved.

Future reprocessing becomes possible.

Ingestion Architecture

The pipeline will use:

			
Tender Sources
       |
       v
Source Connectors
       |
       v
Ingestion Service
       |
       v
Raw Tender Records
       |
       v
Normalized Tenders

		

Responsibilities:

Component	Responsibility
Connector	Download source data
Ingestion Service	Coordinate workflow
Raw Record Repository	Store original payload
Tender Service	Create/update tenders
Processing Error Service	Track failures

New Folder Structure

Create:

			
app/
│
├── ingestion/
│   ├── base_connector.py
│   ├── mock_connector.py
│   ├── registry.py
│   └── __init__.py
│
├── services/
│   ├── ingestion_service.py
│   └── ...
│
├── schemas/
│   ├── ingestion.py
│   └── ...

		

This separates ingestion logic from CRUD services.

Why Start with a Mock Connector?

Real tender portals:

Change frequently
Use different APIs
May require authentication
May impose rate limits

For the MVP:

We first build the ingestion framework.

Then we connect real portals later.

A mock connector allows development without external dependencies.

Creating the Base Connector

Create:

app/ingestion/base_connector.py

			
from abc import ABC
from abc import abstractmethod
class BaseTenderConnector(ABC):
    @abstractmethod
    def fetch_tenders(self):
        pass

		

Every connector must implement:

fetch_tenders()

Why Use an Abstract Base Class?

Without it:

Each connector could behave differently.

With a shared interface:

connector.fetch_tenders()

works consistently for every source.

Examples:

			
TEDConnector
SAMConnector
GovUKConnector
MockConnector

All follow the same contract.

Creating the First Mock Connector

Create:

app/ingestion/mock_connector.py

			
from datetime import datetime
from app.ingestion.base_connector import (
    BaseTenderConnector
)
class MockConnector(
    BaseTenderConnector
):
    def fetch_tenders(self):
        return [
            {
                "external_id": "MOCK-001",
                "title":
                    "Cloud Migration Services",
                "description":
                    "Migration of legacy systems.",
                "buyer_name":
                    "City Council",
                "published_at":
                    datetime.utcnow(),
                "deadline_at":
                    datetime.utcnow(),
                "country_code":
                    "NL",
                "estimated_value":
                    50000,
                "currency":
                    "EUR",
            },
            {
                "external_id": "MOCK-002",
                "title":
                    "Cybersecurity Audit",
                "description":
                    "Independent audit services.",
                "buyer_name":
                    "Regional Authority",
                "published_at":
                    datetime.utcnow(),
                "deadline_at":
                    datetime.utcnow(),
                "country_code":
                    "BE",
                "estimated_value":
                    75000,
                "currency":
                    "EUR",
            },
        ]

		

This simulates a real source.

Building the Connector Registry

Create:

app/ingestion/registry.py

			
from app.ingestion.mock_connector import (
    MockConnector
)
CONNECTOR_MAP = {
    "mock": MockConnector,
}

		

Helper:

			
def get_connector(
    source_type: str
):
    connector_class = (
        CONNECTOR_MAP[source_type]
    )
    return connector_class()

		

Why Use a Registry?

Without a registry:

			
if source_type == "mock":
    ...
elif source_type == "ted":
    ...
elif source_type == "sam":
    ...

		

This becomes difficult to maintain.

A registry allows:

			
connector =
    get_connector(source.source_type)

No conditional logic required.

Creating the Ingestion Run Repository

We already have the model.

Create:

app/repositories/ingestion_run_repository.py

			
from app.models.ingestion_run import (
    IngestionRun
)
from app.repositories.base import (
    BaseRepository
)
class IngestionRunRepository(
    BaseRepository[IngestionRun]
):
    def __init__(self, db):
        super().__init__(
            IngestionRun,
            db
        )

		

Simple for now.

More reporting methods will be added later.

Why Track Ingestion Runs?

Every ingestion execution should be auditable.

Questions we want to answer:

			
When did ingestion run?
How many records were processed?
How many failures occurred?
Which source failed?

The IngestionRun table provides operational visibility.

Creating the Raw Record Repository

Create:

app/repositories/raw_tender_record_repository.py

			
from app.models.raw_tender_record import (
    RawTenderRecord
)
from app.repositories.base import (
    BaseRepository
)
class RawTenderRecordRepository(
    BaseRepository[
        RawTenderRecord
    ]
):
    def __init__(self, db):
        super().__init__(
            RawTenderRecord,
            db
        )

		

Creating the Processing Error Repository

Create:

app/repositories/processing_error_repository.py

			
from app.models.processing_error import (
    ProcessingError
)
from app.repositories.base import (
    BaseRepository
)
class ProcessingErrorRepository(
    BaseRepository[
        ProcessingError
    ]
):
    def __init__(self, db):
        super().__init__(
            ProcessingError,
            db
        )

		

Why Track Processing Errors?

Failures will occur:

			
Network failures
Invalid data
Missing fields
Unexpected formats
API changes

		

Rather than losing failures:

We store them.

This dramatically improves troubleshooting.

Creating the Ingestion Service

Create:

app/services/ingestion_service.py

This service orchestrates the entire workflow.

Service Initialization

			
from sqlalchemy.orm import Session
from app.ingestion.registry import (
    get_connector
)
from app.repositories.ingestion_run_repository import (
    IngestionRunRepository
)
from app.repositories.raw_tender_record_repository import (
    RawTenderRecordRepository
)
from app.repositories.processing_error_repository import (
    ProcessingErrorRepository
)
from app.repositories.tender_source_repository import (
    TenderSourceRepository
)
from app.services.tender_service import (
    TenderService
)

		

Constructor:

			
class IngestionService:
    def __init__(
        self,
        db: Session
    ):
        self.db = db
        self.tender_service = (
            TenderService(db)
        )
        self.source_repository = (
            TenderSourceRepository(db)
        )
        self.ingestion_repository = (
            IngestionRunRepository(db)
        )
        self.raw_repository = (
            RawTenderRecordRepository(db)
        )
        self.error_repository = (
            ProcessingErrorRepository(db)
        )

		

Starting an Ingestion Run

Create:

			
def create_run(
    self,
    source_id
):

Implementation:

			
from app.models.ingestion_run import (
    IngestionRun
)
run = IngestionRun(
    source_id=source_id,
    status="running",
    total_records=0,
    processed_records=0,
    failed_records=0,
)
self.ingestion_repository.add(run)
self.db.commit()
return run

		

Why Create a Run Record First?

If ingestion crashes:

We still know it started.

This helps operational monitoring.

Fetching Records

Inside:

process_source()

Load source:

			
source = (
    self.source_repository.get_by_id(
        source_id
    )
)

		

Load connector:

			
connector = get_connector(
    source.source_type
)

Fetch records:

records = connector.fetch_tenders()

Storing Raw Records

For each record:

			
raw_record = RawTenderRecord(
    source_id=source.id,
    external_id=record[
        "external_id"
    ],
    payload=record,
)

		

Store:

			
self.raw_repository.add(
    raw_record
)

Raw data is now preserved.

Why Store Before Normalization?

Suppose normalization crashes:

Raw data still exists.

The record can be reprocessed later.

Duplicate Detection

Before creating a tender:

			
existing = (
    self.tender_service
    .repository
    .get_by_source_external_id(
        source_id=source.id,
        external_id=record[
            "external_id"
        ]
    )
)

		

If found:

Update existing record

Otherwise:

Create new tender

Creating a Tender

Use the existing service layer:

			
self.tender_service.create_tender(
    payload
)

This ensures:

Validation
Duplicate protection
Transaction handling

remain centralized.

Why Reuse the Service Layer?

Bad:

Duplicate tender creation logic

Good:

TenderService.create_tender()

Single source of truth.

Tracking Statistics

Update counters:

			
run.total_records += 1
run.processed_records += 1

On failure:

run.failed_records += 1

These values become dashboard metrics.

Logging Processing Errors

Inside exception handling:

			
from app.models.processing_error import (
    ProcessingError
)

Create:

			
error = ProcessingError(
    source_id=source.id,
    external_id=record.get(
        "external_id"
    ),
    error_message=str(exc),
)

		

Store:

			
self.error_repository.add(
    error
)

Commit:

self.db.commit()

Why Continue After Failures?

Suppose:

			
records
bad record

Stopping the entire run would be wasteful.

Instead:

			
succeed
logged as failure

The pipeline remains resilient.

Completing the Run

After processing:

run.status = "completed"

Set:

			
run.completed_at =
    datetime.utcnow()

Commit:

self.db.commit()

Handling Run-Level Failures

Wrap the pipeline:

			
try:
    ...
except Exception:

Set:

run.status = "failed"

Commit:

self.db.commit()

This distinguishes:

			
Completed
Failed
Running

states.

Creating an Ingestion Schema

Create:

app/schemas/ingestion.py

			
from pydantic import BaseModel
class IngestionResult(
    BaseModel
):
    run_id: str
    total_records: int
    processed_records: int
    failed_records: int
    status: str

		

This provides API responses later.

Testing the Mock Pipeline

Create:

tests/services/test_ingestion_service.py

Example:

			
def test_mock_ingestion(
    db_session
):

Run:

			
service = IngestionService(
    db_session
)

Execute:

			
result =
    service.process_source(
        source_id
    )

Assertions:

			
assert (
    result.total_records
    == 2
)
assert (
    result.failed_records
    == 0
)

		

Why Test the Entire Pipeline?

Unit tests verify:

			
Repository logic
Service logic
Validation

Pipeline tests verify:

End-to-end ingestion

including:

Connector
Storage
Normalization
Statistics

Future Connectors

After the MVP framework works:

			
TED (European Union)
SAM.gov (United States)
Contracts Finder (UK)
Mercell
Public Procurement Portals

		

can be added.

Each becomes:

			
class TEDConnector(
    BaseTenderConnector
)

No ingestion-service changes required.

Current Architecture

The backend now supports:

			
Tender Sources
      |
      v
Connectors
      |
      v
Ingestion Service
      |
      +--> Raw Records
      +--> Tenders
      +--> Errors
      +--> Statistics
      |
      v
PostgreSQL

		

This is the operational core of the platform.

Current Deliverables

At this stage we now have:

			
✔ Connector architecture
✔ Base connector abstraction
✔ Mock tender connector
✔ Connector registry
✔ Ingestion run tracking
✔ Raw record storage
✔ Duplicate detection
✔ Tender creation workflow
✔ Error logging
✔ Processing statistics
✔ Pipeline testing strategy

		

The AI Tender Opportunity Scanner can now ingest tender opportunities into the platform.

Recommended Git Commit

			
git add .
git commit -m "Implement tender ingestion pipeline"

What We Will Build Next

The platform can now collect tenders.

The next step is transforming raw tender data into useful intelligence.

We will begin building:

AI Analysis Pipeline

This pipeline will:

Generate tender summaries
Extract requirements
Identify technologies
Detect industries
Estimate opportunity relevance
Prepare tender embeddings

The next article in the series is:

“Building the AI Analysis Pipeline for Tender Summarization and Information Extraction.”

Conclusion

The Tender Ingestion Pipeline is the first truly operational subsystem of the AI Tender Opportunity Scanner.

The platform can now retrieve external tender data, preserve raw records, normalize information, track ingestion activity, and recover gracefully from failures.

This foundation ensures that future AI analysis and matching capabilities will operate on a reliable, auditable, and continuously updated stream of procurement opportunities.

With ingestion complete, we are finally ready to introduce AI into the platform and begin transforming tender data into actionable business intelligence.

Building the Tender Ingestion Pipeline for the AI Tender Opportunity Scanner

Introduction

Development Progress

What We Are Building

Why Preserve Raw Data?

Ingestion Architecture

New Folder Structure

Why Start with a Mock Connector?

Creating the Base Connector

Why Use an Abstract Base Class?

Creating the First Mock Connector

Building the Connector Registry

Why Use a Registry?

Creating the Ingestion Run Repository

Why Track Ingestion Runs?

Creating the Raw Record Repository

Creating the Processing Error Repository

Why Track Processing Errors?

Creating the Ingestion Service

Service Initialization

Starting an Ingestion Run

Why Create a Run Record First?

Fetching Records

Storing Raw Records

Why Store Before Normalization?

Duplicate Detection

Creating a Tender

Why Reuse the Service Layer?

Tracking Statistics

Logging Processing Errors

Why Continue After Failures?

Completing the Run

Handling Run-Level Failures

Creating an Ingestion Schema

Testing the Mock Pipeline

Why Test the Entire Pipeline?

Future Connectors

Current Architecture

Current Deliverables

Recommended Git Commit

What We Will Build Next

Conclusion

Share this:

Like this:

Discover more from BidRadar