Scraping Discipline & Extraction Framework Specification

A lightweight, zero-dependency data validation, normalizer stacking, multi-extractor candidate execution, and diagnostic provenance engine designed specifically for web scraping pipelines.

---

1. Executive Summary & Core Objective

In automated data extraction, traditional contract-validation libraries (such as standard Pydantic, Attrs, or Marshmallow) introduce substantial inefficiency due to their fail-fast design. When extracting unstructured web content across thousands of pages, throwing fatal validation errors over minor anomalies leads to discarded records, wasted network bandwidth, unnecessary crawler footprint, and inflated compute costs.

This framework introduces a discipline-first, multi-extractor, extraction-agnostic architecture to maximize usable data yield while preserving strict auditability and data integrity:

• Separation of Extraction vs. Discipline: Decouples extraction mechanisms (BeautifulSoup, Playwright locators, Scrapy selectors, regex, direct dict lookups, or heuristic inference engines) from data normalization, type coercion, and boundary disciplines.
• Multi-Extractor Execution (extractors: Sequence[...]): Enables multi-perspective extraction hooks across layout variations and engines to produce comprehensive candidate profile arrays without premature single-winner discard.
• Cascading Normalization Stacks: Composable schema pre-processors stacked deterministically with field-specific transformers.
• Modern Configuration & Native Type Support: Full support for model_config: ConfigDict, standard Python types (int, float, str, Optional, Union), and typing.Annotated discipline bindings.
• Parameterized Type Factories & Semantic Dispatch: Reusable type builders like Number(min=..., max=...) dispatching semantic presets such as PositiveNumber and Percentage.
• Declarative Outlier & Missing Value Policies: Granular handling of boundary violations (WARN, CLAMP, DEFAULT, QUARANTINE, ERROR) and missing fields (USE_DEFAULT, NULLIFY, ERROR) to eliminate unhandled crawl terminations.
• Provenance & Diagnostic Tracking: Automatic per-field metadata (raw, source, confidence, outlier, issues, formatted) for total audit transparency.
• Total Parsing Assurance: Every parser operates as a total function, returning structured outcomes without throwing unhandled exceptions.

---

2. Architectural Pipeline

Raw Source Context (DOM Tree, Page Locator, Dict, or Inference Engine)
   │
   ▼
[Stage 1: Multi-Extractor Execution] (`extractors=[ext1, ext2, ext3]`)
   │  (Executes all extractors across fields to produce extraction candidate profiles)
   │
   ▼
[Stage 2: Schema Normalizer Stack] (`model_config["normalizers"]`)
   │  (e.g., whitespace collapsing, unicode cleanup, entity unescaping)
   │
   ▼
[Stage 3: Local Field Normalizers] (`Field(normalizers=[...])`)
   │  (e.g., Arabic-to-ASCII digit conversion, currency regex, unit stripping)
   │
   ▼
[Stage 4: Semantic Type Coercion & Discipline] (`received_projects: PositiveNumber = Field(max=60)`)
   │  (Safe type casting, min/max constraint evaluation)
   │
   ▼
[Stage 5: Outlier Policy Evaluation] (`on_outlier`)
   │  (WARN, CLAMP, DEFAULT, ERROR, QUARANTINE)
   │
   ▼
[Stage 6: Record-Level Invariants & Coherence Rules] (`@coherence`)
   │  (Cross-field validation, record quality classification per profile outcome)
   │
   ▼
Array of Extracted Profiles + FieldMeta Provenance Reports (`profiles: List[ParsedResult]`)

---

3. Declarative Schema & Canonical Field API

Field(...) is declared as the canonical default and primary standard API for all standard scraping schemas, tutorials, and quickstart documentation. It serves as a high-level ergonomic factory that compiles declarative arguments into the framework's internal Discipline and Extract engine representations.

3.1 Field Disciplines & Modern Schema Definition (Canonical Quickstart)

from enum import Enum
from typing import Any, Callable, List, Optional, Sequence, Union, Annotated, TypedDict
from scrape_core import (
    Schema, Field, Number, PositiveNumber, Percentage, Duration,
    Discipline, OutlierPolicy, FallbackPolicy, coherence, MetricQuality,
    ConfigDict, RawCandidate, ParsedResult
)

class FreelancerProfile(Schema):
    # Modern TypedDict schema configuration
    model_config: ConfigDict = {
        "normalizers": [
            lambda text: text.strip() if isinstance(text, str) else text,
            lambda text: " ".join(text.split()) if isinstance(text, str) else text,
        ],
        "default_outlier_policy": OutlierPolicy.WARN,
        "arabic_digit_translation": True,
    }

    # Multi-extractor pipeline with semantic type annotation and field bounds
    received_projects: PositiveNumber = Field(
        max=60,
        on_outlier=OutlierPolicy.WARN,
        normalizers=[
            lambda val: val.replace("مشروع", "").strip() if isinstance(val, str) else val
        ],
        extractors=[
            lambda soup: soup.select_one(".received-projects"),
            lambda soup: soup.find("div", {"data-metric": "received"}),
            lambda ctx: ctx.infer_engine.extract("received_projects"),
        ],
        default=0,
    )

    # Duration with boundary clamping and multi-engine extractors
    avg_response_time_minutes: Duration = Field(
        unit="minutes", min=1.0, max=10080.0,  # Max 7 days
        on_outlier=OutlierPolicy.CLAMP,
        fallback=FallbackPolicy.USE_DEFAULT,
        extractors=[
            lambda ctx: ctx.infer_engine.extract("avg_response_time_raw"),
            lambda soup: soup.select_one(".response-time"),
        ],
        default=1440.0,
    )

    # Percentage field with local string cleanup
    completion_rate: Percentage = Field(
        min=0.0, max=100.0,
        on_outlier=OutlierPolicy.CLAMP,
        normalizers=[
            lambda v: v.rstrip("%").strip() if isinstance(v, str) else v
        ],
        extractors=[
            lambda soup: soup.select_one(".completion-rate-badge"),
            lambda soup: soup.find("span", class_="rate"),
        ],
        default=0.0,
    )

    # Bounded Number and Nullable support with direct selector extractor
    rating: Optional[Number] = Field(
        min=0.0, max=5.0,
        on_outlier=OutlierPolicy.CLAMP,
        extractors=[lambda soup: soup.select_one(".rating-badge")],
        default=None,
    )

    # Dynamic callable default (e.g. timestamp or container factory)
    scraped_at: str = Field(
        extractors=[lambda soup: soup.find("meta", {"name": "scraped-at"})],
        default=lambda: datetime.utcnow().isoformat(),
    )

    tags: List[str] = Field(
        extractors=[lambda soup: [t.get_text(strip=True) for t in soup.select(".skill-tag")]],
        default=list,
    )

    # Cross-field coherence rule
    @coherence
    def validate_project_stats(self, report):
        if self.received_projects == 0 and self.completion_rate > 0.0:
            report.add_issue(
                field="completion_rate",
                issue="completion_rate_with_zero_projects",
                severity=MetricQuality.SUSPECT,
                message="Non-zero completion rate reported for freelancer with zero received projects."
            )

---

4. Multi-Extractor Execution & Candidate Extraction

Rather than forcing a single winner or guessing which extraction strategy is "most accurate", the framework executes all declared extractors. Each extractor is evaluated to produce its own parsed candidate values and metadata, allowing downstream consumers to inspect, compare, or reconcile all extraction perspectives.

4.1 Standardized Return Envelope (RawCandidate)

Extractors may return raw primitives (str, Tag, None) or an explicit RawCandidate carrying provenance and confidence metadata:

from dataclasses import dataclass
from typing import Any, Optional, Sequence, Union, Callable, List

@dataclass(frozen=True)
class RawCandidate:
    value: Any
    source: str = "custom_extractor"
    confidence: float = 1.0

ExtractorCallable = Callable[[Any], Union[Any, RawCandidate, None]]

def execute_all_extractors(
    context: Any, 
    extractors: Sequence[ExtractorCallable]
) -> List[RawCandidate]:
    """Executes all extractors in sequence, collecting all non-empty raw candidates."""
    candidates = []
    for idx, ext in enumerate(extractors):
        try:
            res = ext(context)
            if res is None:
                continue
            
            # Handle explicit RawCandidate instances
            if isinstance(res, RawCandidate):
                if res.value not in (None, "", "لم يحسب بعد", "n/a"):
                    candidates.append(res)
            else:
                # Handle DOM elements or raw scalar values
                raw_str = res.get_text() if hasattr(res, "get_text") else str(res)
                if raw_str.strip() not in ("", "None", "null", "n/a", "لم يحسب بعد"):
                    candidates.append(
                        RawCandidate(
                            value=res, 
                            source=getattr(ext, "__name__", f"extractor_{idx}"),
                            confidence=max(0.2, 1.0 - (idx * 0.15))
                        )
                    )
        except Exception:
            continue
            
    return candidates

---

5. Parameterized Type Factory & Type Definition Styles

Rather than maintaining isolated subclasses for each numeric constraint, a unified Number factory provides parameterization, type annotation dispatch, and seamless integration with Field(...) keyword arguments:

from typing import Optional, Any
from scrape_core import ParseOutcome

class Number:
    """Base numeric discipline factory and type descriptor."""
    def __init__(
        self, 
        min: Optional[float] = None, 
        max: Optional[float] = None, 
        dtype: type = float
    ):
        self.min = min
        self.max = max
        self.dtype = dtype

    def __call__(self, **kwargs) -> "Number":
        """Refines parameters fluently or creates specialized instances."""
        new_params = {"min": self.min, "max": self.max, "dtype": self.dtype}
        new_params.update(kwargs)
        return Number(**new_params)

    def parse(self, raw: Any) -> ParseOutcome:
        # Safe total parsing implementation with bounds checking
        ...

# Semantic type annotations for fields
PositiveNumber = Number(min=0, dtype=int)
PositiveFloat  = Number(min=0.0, dtype=float)
Percentage     = Number(min=0.0, max=100.0, dtype=float)

5.1 Supported Type Definition Styles

The framework supports three flexible declaration styles, with Style A being the recommended convention:

# Style A: Static Type Hint Annotation (Recommended - Cleanest for IDEs & Linters)
received_projects: PositiveNumber = Field(max=60, default=0)

# Style B: Explicit type= parameter (Great for dynamic schema factories & unannotated code)
received_projects = Field(type=PositiveNumber(max=60), default=0)

# Style C: Standard Python Type with Explicit Discipline
received_projects: int = Field(type=PositiveNumber(max=60), default=0)

5.2 Direct Semantic Annotation with Field(...) Parameter Overrides

When declaring schema attributes using Style A, developers use the semantic type directly as the type annotation while supplying field-specific bounds and behaviors directly inside Field(...):

# Clean and natural syntax:
received_projects: PositiveNumber = Field(max=60, default=0)
avg_response_time: Duration = Field(unit="minutes", max=10080.0, default=1440.0)
completion_rate: Percentage = Field(default=0.0)
rating: Optional[Number] = Field(min=0.0, max=5.0, default=None)

5.3 Dynamic & Callable Defaults (default: Union[Any, Callable[[], Any]])

The default parameter accepts both static scalar values and zero-argument callables/factories (such as functions, lambdas, or type constructors like list, dict, or datetime.utcnow). The framework also supports default_factory as an explicit parameter alias.

from datetime import datetime
import uuid

class ProjectRecord(Schema):
    # Static scalar default
    status: str = Field(default="pending")

    # Callable factory for fresh collections (prevents mutable default bugs)
    tags: List[str] = Field(default=list)
    metadata: Dict[str, Any] = Field(default_factory=dict)

    # Dynamic value generation evaluated at parse time
    scraped_at: str = Field(default=lambda: datetime.utcnow().isoformat())
    trace_id: str = Field(default=lambda: str(uuid.uuid4()))

Default Resolution Mechanism

When a default is required (due to missing values, failed extraction tiers, or OutlierPolicy.DEFAULT), the framework evaluates the default dynamically:

def resolve_default(default_spec: Union[Any, Callable[[], Any]]) -> Any:
    """Evaluates callable defaults or returns static values safely."""
    if callable(default_spec):
        return default_spec()
    return default_spec

This guarantees:

1. Zero Shared Mutable State: Collection factories (list, dict) instantiate a new object per parsed record.
2. Dynamic Generation: Timestamps, session tokens, or UUIDs are computed at the exact moment of extraction fallback.

---

6. Native Python Types, Union, and Annotated Support

The framework integrates seamlessly with modern Python type hinting:

6.1 Annotated Discipline Binding

from typing import Annotated, Optional

# Parameterized discipline definitions
ProjectsCount = Annotated[int, Discipline(max=60, on_outlier=OutlierPolicy.WARN)]
ResponseDuration = Annotated[float, Discipline(unit="minutes", max=10080, on_outlier=OutlierPolicy.CLAMP)]

class FreelancerProfile(Schema):
    received_projects: ProjectsCount = Field(
        extractors=[
            lambda soup: soup.select_one(".received-projects"),
            lambda ctx: ctx.infer_engine.extract("received_projects"),
        ],
        default=0
    )

    rating: Optional[float] = Field(
        extractors=[lambda s: s.select_one(".rating-badge")],
        default=None
    )

---

7. Modern Configuration Pattern (ConfigDict)

Replacing inner classes with typed dictionary structures ensures type validation, editor autocompletion, and schema immutability:

from typing import TypedDict, List, Callable, Optional
from scrape_core import OutlierPolicy

class ConfigDict(TypedDict, total=False):
    normalizers: List[Callable[[Any], Any]]
    default_outlier_policy: OutlierPolicy
    auto_strip_text: bool
    arabic_digit_translation: bool
    strict_types: bool

---

8. Normalizer Pipeline Hierarchy & Stacking

Normalizers are pure, composable callables applied in deterministic sequence:

$$\text{Pipeline}(x) = (\text{LocalNormalizers} \circ \text{SchemaNormalizers})(x)$$

1. Schema-Level Normalizers: Configured in model_config["normalizers"]. Executed across all extracted inputs before local transformations.
2. Local Field Normalizers: Declared per field via Field(normalizers=[...]). Run after schema-level normalizers.
3. Override Flag: Setting Field(override_normalizers=True) bypasses schema-level processing for specialized fields.

---

9. Outlier Policies & Missing Value Handling

9.1 Outlier Policies (OutlierPolicy)

Policy	Action Taken	Diagnostic Result	Use Case
WARN	Retains parsed value as-is.	FieldMeta.outlier = True, issues += ["outlier_warning"]	Preserving raw observations for audit or downstream research.
CLAMP	Truncates value to declared $[min, max]$.	FieldMeta.value = bound, issues += ["outlier_clamped"]	Percentages, ratings, and bounded natural metrics.
DEFAULT	Discards out-of-bounds value; assigns field.default.	FieldMeta.value = default, issues += ["outlier_defaulted"]	Corrupted metrics that would distort analytical aggregations.
QUARANTINE	Retains value; marks record quality as SUSPECT.	RecordReport.quality = MetricQuality.QUARANTINE	High-impact discrepancies requiring human inspection.
ERROR	Raises FieldValidationError immediately.	Aborts schema execution.	Strict security constraints or non-negotiable primary keys.

9.2 Missing Value Policies (FallbackPolicy)

When an extractor fails or yields empty/placeholder content for a specific field:

• USE_DEFAULT: Substitutes field.default and records diagnostic metadata.
• NULLIFY: Sets field value to None.
• ERROR: Raises an error if the field is mandatory for that extractor tier.

---

10. Multi-Profile Provenance & Runtime Inspection

Because all extractors are executed unconditionally, Schema.parse() returns an array/list of ParsedResult containers (one for each extraction tier or candidate set), pairing the typed model instance with its full provenance and coherence report:

@dataclass(frozen=True)
class FieldMeta:
    value: Any
    raw: str
    source: str
    confidence: float
    outlier: bool
    issues: List[str]
    type: str
    formatted: str

@dataclass(frozen=True)
class ParsedResult:
    profile: FreelancerProfile
    report: Dict[str, FieldMeta]
    quality: MetricQuality
    extractor_source: str

10.1 Runtime Inspection Example

# Returns an array/list of extracted candidate results across all declared extractors
results: List[ParsedResult] = FreelancerProfile.parse(html_document)

for idx, result in enumerate(results):
    profile = result.profile
    report = result.report
    
    print(f"=== Candidate Profile from {result.extractor_source} ===")
    print(f"Received projects: {profile.received_projects}")
    print(f"Avg response time: {profile.avg_response_time_minutes}")
    
    # Granular per-field provenance for this extraction candidate
    meta = report["received_projects"]
    print(f"Raw text: {meta.raw}")               # e.g., "٤٥ مشروع"
    print(f"Confidence: {meta.confidence}")       # 1.0
    print(f"Outlier flagged: {meta.outlier}")     # False
    print(f"Field issues: {meta.issues}")         # []
    print(f"Inferred type: {meta.type}")          # "PositiveNumber"
    
    # Candidate-level coherence quality
    print(f"Record Quality: {result.quality}")    # MetricQuality.OK

---

11. Outlier Report States & Downstream Pipeline Utilization

The FieldMeta diagnostic report converts extraction anomalies from passive debug logs into active operational assets across downstream ingestion, monitoring, and analytical systems.

11.1 Concrete Visualizations of Report States Per Case

Given a schema field definition:

received_projects: PositiveNumber = Field(min=0, max=60, default=0)

Case A: Nominal Value (Clean Extraction)

• Raw Input: " 24 مشروع "
• Parsed Value: 24
• Report State:

FieldMeta(
    value=24,
    raw="  24 مشروع  ",
    source="extractor_0",
    confidence=1.0,
    outlier=False,
    issues=[],
    type="PositiveNumber",
    formatted="24"
)
# Record Quality: MetricQuality.OK

Case B: Outlier with OutlierPolicy.WARN

• Raw Input: " 140 مشروع " (Exceeds max=60)
• Parsed Value: 140 (Preserved as-is)
• Report State:

FieldMeta(
    value=140,
    raw=" 140 مشروع ",
    source="extractor_0",
    confidence=0.75,
    outlier=True,
    issues=["above_max", "outlier_warning"],
    type="PositiveNumber",
    formatted="140"
)
# Record Quality: MetricQuality.OK

Case C: Outlier with OutlierPolicy.CLAMP

• Raw Input: " 150% " for completion_rate: Percentage = Field(max=100.0)
• Parsed Value: 100.0 (Truncated to boundary)
• Report State:

FieldMeta(
    value=100.0,
    raw=" 150% ",
    source="extractor_0",
    confidence=0.80,
    outlier=True,
    issues=["above_max", "outlier_clamped"],
    type="Percentage",
    formatted="100.0%"
)
# Record Quality: MetricQuality.OK

Case D: Outlier with OutlierPolicy.DEFAULT

• Raw Input: " -999 " (Corrupted or negative count)
• Parsed Value: 0 (Safe fallback substituted)
• Report State:

FieldMeta(
    value=0,
    raw=" -999 ",
    source="extractor_0",
    confidence=0.10,
    outlier=True,
    issues=["below_min", "outlier_defaulted"],
    type="PositiveNumber",
    formatted="0"
)
# Record Quality: MetricQuality.SUSPECT

Case E: Outlier with OutlierPolicy.QUARANTINE

• Raw Input: " 95000 " (Extreme value signaling layout shift or parser misalignment)
• Parsed Value: 95000
• Report State:

FieldMeta(
    value=95000,
    raw=" 95000 ",
    source="extractor_0",
    confidence=0.30,
    outlier=True,
    issues=["above_max", "quarantine_flagged"],
    type="PositiveNumber",
    formatted="95000"
)
# Record Quality: MetricQuality.QUARANTINE

---

11.2 Downstream Pipeline Utilization Patterns

ParsedResult (Profile + Report)
   │
   ├─► 1. Data Warehousing & Partitioning (ELT / Ingestion Routing)
   ├─► 2. Downstream Statistical Weighting & Aggregation
   ├─► 3. Real-Time Crawler Health & Layout Drift Detection
   ├─► 4. Multi-Extractor Cross-Reconciliation & Scoring
   └─► 5. Human-in-the-Loop Audit & Triage Dashboards

1. Ingestion Routing & Dead-Letter Queues (DLQ):

   • Records with MetricQuality.OK and outlier == False route directly to production tables.
   • Records with outlier == True (clamped/defaulted) are ingested with an is_imputed: True flag.
   • Records with MetricQuality.QUARANTINE route to a staging DLQ for investigation.

2. Downstream Statistical Weighting & Clean Analytics:

   • Aggregation queries can filter out imputed metrics (WHERE received_projects_outlier = FALSE) to prevent distorting sensitive statistics (medians, standard deviations).
   • Historical queries can be re-run against FieldMeta.raw if business bounds change.

3. Real-Time Crawler Health & Layout Drift Telemetry:

   • Tracks anomaly rates across crawl jobs: $$\text{Anomaly Rate} = \frac{\sum \text{outlier_issues}}{\text{Total Scraped Records}}$$
   • If the anomaly rate exceeds a threshold (e.g. $> 15%$), automated alerts trigger before corrupted records spread.

4. Multi-Extractor Reconciliation & Candidate Selection:

   • Downstream arbiters compare candidates across ParsedResult entries using issue counts and confidence scores to pick the highest-fidelity profile.

5. Human-in-the-Loop Audit & Debug Dashboards:

   • Dashboards color-code fields by quality and display original FieldMeta.raw text and applied policies on hover for complete transparency.

---

12. Framework Interoperability Layer: Discipline & Extract Internal Engine

While Field(...) is the canonical default syntax for standalone schemas and quickstarts, Discipline and Extract constitute the framework's internal core engine and framework interoperability extension layer.

This layer enables developers working with existing ecosystems (pydantic.BaseModel, sqlmodel.SQLModel, or plain dataclasses) to attach scraping disciplines directly using standard Python typing.Annotated metadata without name collisions against Pydantic's native Field or schema duplication:

                      ┌───────────────────────────────────────┐
                      │    typing.Annotated[T, Discipline]    │
                      │   (Standard Library Metadata Marker)  │
                      └──────────────────┬────────────────────┘
                                         │
        ┌────────────────────────────────┼────────────────────────────────┐
        ▼                                ▼                                ▼
  [Plain Class / BS4]           [Pydantic / FastAPI]             [SQLModel / SQLAlchemy]
  (Zero dependencies,            (Native API schema,             (Direct ORM database
   lightweight scripts)           FastAPI validation)             table persistence)

12.1 Universal Discipline & Extractor Markers

from dataclasses import dataclass
from typing import Any, Callable, Sequence, Union, Optional
from scrape_core import OutlierPolicy

@dataclass(frozen=True)
class Discipline:
    min: Optional[float] = None
    max: Optional[float] = None
    unit: Optional[str] = None
    on_outlier: OutlierPolicy = OutlierPolicy.WARN
    normalizers: Sequence[Callable[[Any], Any]] = ()

@dataclass(frozen=True)
class Extract:
    extractors: Sequence[Callable[[Any], Any]]
    override_normalizers: bool = False

---

12.2 Scenario 1: Raw Script Mode (Standard dataclass or Plain Class)

Zero dependencies required—pure standard library:

from dataclasses import dataclass
from typing import Annotated, Optional
from bs4 import BeautifulSoup
from scrape_core import Discipline, Extract, OutlierPolicy, parse_item

@dataclass
class SimpleFreelancer:
    received_projects: Annotated[
        int,
        Discipline(min=0, max=60, on_outlier=OutlierPolicy.WARN),
        Extract([lambda soup: soup.select_one(".received-projects")]),
    ] = 0

    name: Annotated[
        str,
        Extract([lambda soup: soup.select_one(".name")]),
    ] = ""

# Parse directly with BeautifulSoup
soup = BeautifulSoup(html_content, "html.parser")
results = parse_item(SimpleFreelancer, soup)

print(results[0].profile.received_projects)  # 24
print(results[0].report["received_projects"].confidence)  # 1.0

---

12.3 Scenario 2: Pydantic V2 / FastAPI Integration

Direct integration with pydantic.BaseModel without duplicating field definitions:

from typing import Annotated, Optional
from pydantic import BaseModel, Field as PydanticField
from scrape_core import Discipline, Extract, OutlierPolicy, parse_item

class FreelancerDTO(BaseModel):
    # Pydantic handles API validation; Annotated attaches scraping disciplines
    received_projects: Annotated[
        int,
        Discipline(min=0, max=60, on_outlier=OutlierPolicy.WARN),
        Extract([
            lambda soup: soup.select_one(".received-projects"),
            lambda ctx: ctx.infer_engine.extract("received_projects"),
        ]),
    ] = PydanticField(default=0, description="Number of received projects")

    name: Annotated[
        str,
        Extract([lambda soup: soup.select_one(".name")]),
    ] = PydanticField(default="", max_length=150)

# 1. Used as Scraping Parser:
results = parse_item(FreelancerDTO, raw_html)
freelancer_pydantic_instance = results[0].profile  # Valid Pydantic instance

# 2. Used directly in FastAPI routes without re-declaring models:
# @app.post("/freelancers", response_model=FreelancerDTO)

---

12.4 Scenario 3: SQLModel / Database Table Model

One single model for Scraping, API, and Database ORM:

from typing import Annotated, Optional
from sqlmodel import SQLModel, Field as SQLField
from scrape_core import Discipline, Extract, OutlierPolicy, parse_item

class FreelancerRecord(SQLModel, table=True):
    id: Optional[int] = SQLField(default=None, primary_key=True)

    received_projects: Annotated[
        int,
        Discipline(min=0, max=60, on_outlier=OutlierPolicy.WARN),
        Extract([lambda soup: soup.select_one(".received-projects")]),
    ] = SQLField(default=0, index=True)

    name: Annotated[
        str,
        Extract([lambda soup: soup.select_one(".name")]),
    ] = SQLField(default="", max_length=150)

# Scrape -> Ingest into DB directly:
results = parse_item(FreelancerRecord, html)
with Session(engine) as session:
    session.add(results[0].profile)
    session.commit()

---

12.5 Universal Introspection Engine (parse_item)

The standalone engine inspects type annotations via typing.get_type_hints(cls, include_extras=True):

from typing import get_type_hints, get_origin, get_args, Any, List

def parse_item(target_cls: type, context: Any) -> List[ParsedResult]:
    """Universal parser that extracts metadata from any Annotated class."""
    hints = get_type_hints(target_cls, include_extras=True)
    field_specs = {}

    for field_name, hint in hints.items():
        if get_origin(hint) is Annotated:
            base_type, *metadata = get_args(hint)
            discipline = next((m for m in metadata if isinstance(m, Discipline)), None)
            extract = next((m for m in metadata if isinstance(m, Extract)), None)
            field_specs[field_name] = (base_type, discipline, extract)

    # Executes extractors -> normalizers -> discipline constraints -> instantiates target_cls
    ...

---

13. Practical Efficiency & Yield Benefits

1. Maximized Scraping Yield: Evaluating all extractors unconditionally captures multiple data perspectives without arbitrary single-winner discard, preserving critical information for downstream analytics and reconciliation.
2. Reduced Engineering Overhead: Centralized normalization and parameterized type factories eliminate repetitive per-field parsing logic.
3. End-to-End Diagnostic Traceability: Retention of FieldMeta allows downstream pipelines (DuckDB, Parquet, dashboards) to filter, weight, or merge records by data confidence without losing raw context.