pydantic-core 2.46.4: pydantic_core

# pydantic_core Primitives, URLs, Timezones, and Schema Validation

## Prior Solution Found
`search_known_solution` returned `NO_SAFE_MATCH` for `verify pydantic_core in pkg:pypi/pydantic-core@2.46.4`.

This sample proves key primitives and validation features in `pydantic_core`:
- `pydantic_core.Url`: Parsing RFC-compliant URLs into scheme, authentication credentials, host, port, path, and parsed query parameter lists.
- `pydantic_core.MultiHostUrl`: Multi-host connection string decomposition with individual host and port mapping.
- `pydantic_core.TzInfo`: Fast Rust-backed fixed UTC offset timezone definitions compatible with standard library `datetime`.
- `pydantic_core.to_jsonable_python`: Converting Python types (e.g. datetimes, sets, tuples) into standard JSON-compatible data structures.
- `pydantic_core.PydanticKnownError`: Structured errors with predefined error codes, message formatting templates, and context attributes.
- `pydantic_core.core_schema.tagged_union_schema`: Discriminated union schemas evaluated with `pydantic_core.SchemaValidator`, verifying valid tags and emitting `union_tag_invalid` on unknown discriminators.
- `pydantic_core.core_schema.chain_schema`: Chained sequential validators and transformation functions.

## What a Model Would Have Written
A model attempting to use `pydantic_core` directly often confuses its lower-level Rust primitives with high-level `pydantic` V2 abstractions:
- Trying to import or instantiate `BaseModel` from `pydantic_core` (it resides in `pydantic`).
- Assuming `MultiHostUrl` exposes single-host attributes like `url.host` or `url.username` directly, rather than decomposing endpoints via `url.hosts()`.
- Trying to instantiate standard Python `timezone` instead of leveraging `TzInfo` for C-speed timezone offset representation.
- Forgetting that `core_schema.tagged_union_schema` requires explicit dictionary choices and emits `union_tag_invalid` rather than a generic validation exception on unknown tags.

## How the Wrong Version Fails
- `AttributeError: module 'pydantic_core' has no attribute 'BaseModel'` or `AttributeError: 'MultiHostUrl' object has no attribute 'username'`.
- `ValidationError: 1 validation error for tagged-union: Input should be ... [type=union_tag_invalid, ...]`.
- `TypeError` when passing unexpected schema options to `core_schema` helper builders.

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify pydantic_core in pkg:pypi/pydantic-core@2.46.4
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/pydantic-core@2.46.4
Demonstrate these symbols/APIs:
  - pydantic_core

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:8e5d5f6164ac947f8e2e79278f6a296b5cd4ff1044e09ff54db71d05096f9fd6","contract":["pydantic_core.__version__ matches pinned version 2.46.4.","pydantic_core.Url parses structured URLs and extracts components including scheme, host, port, path, and query parameters.","pydantic_core.MultiHostUrl parses multi-host connection strings into individual host and port mapping definitions.","pydantic_core.TzInfo defines fixed UTC timezone offsets for standard library datetime instances.","pydantic_core.to_jsonable_python converts datetime, set, and tuple objects into JSON-compatible serializable Python structures.","pydantic_core.PydanticKnownError constructs standardized error instances with predefined message templates and error contexts.","pydantic_core.SchemaValidator with tagged_union_schema resolves matching schemas by discriminator tag and reports union_tag_invalid on unknown tags.","pydantic_core.SchemaValidator with chain_schema validates and transforms input through sequential core schema steps."],"goal":"verify pydantic_core in pkg:pypi/pydantic-core@2.46.4","kind":"HOW","packages":["pkg:pypi/pydantic-core@2.46.4"],"schemaVersion":1,"symbols":["pydantic_core"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/pydantic-core@2.46.4"],"schemaVersion":1,"subject":"pkg:pypi/pydantic-core@2.46.4","symbols":["pydantic_core"],"verifierAdapter":"python@1"}

pydantic-core==2.46.4
typing-extensions==4.16.0

{
  "schemaVersion": 1,
  "goal": "verify pydantic_core in pkg:pypi/pydantic-core@2.46.4",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/pydantic-core@2.46.4"
  ],
  "symbols": [
    "pydantic_core"
  ]
}

from datetime import datetime
from typing import Any
from pydantic_core import (
    MultiHostUrl,
    SchemaValidator,
    TzInfo,
    Url,
    core_schema,
    to_jsonable_python,
)


def parse_endpoint(url_str: str) -> dict[str, Any]:
    """Parse a URL into constituent components."""
    url = Url(url_str)
    return {
        "scheme": url.scheme,
        "host": url.host,
        "port": url.port,
        "path": url.path,
        "query": url.query,
        "fragment": url.fragment,
        "query_params": dict(url.query_params()),
    }


def parse_cluster_hosts(url_str: str) -> list[tuple[str | None, int | None]]:
    """Parse a multi-host connection URL into host and port tuples."""
    mhu = MultiHostUrl(url_str)
    return [(h.get("host"), h.get("port")) for h in mhu.hosts()]


def create_offset_datetime(
    year: int, month: int, day: int, hour: int, minute: int, second: int, offset_seconds: int = 0
) -> datetime:
    """Create a datetime instance using TzInfo timezone offset."""
    tz = TzInfo(offset_seconds)
    return datetime(year, month, day, hour, minute, second, tzinfo=tz)


def convert_to_jsonable(obj: Any) -> Any:
    """Convert arbitrary Python types to JSON-compatible data structures."""
    return to_jsonable_python(obj)


def build_tagged_storage_validator() -> SchemaValidator:
    """Build a SchemaValidator using tagged_union_schema with a discriminator."""
    disk_schema = core_schema.typed_dict_schema({
        "type": core_schema.typed_dict_field(
            core_schema.with_default_schema(core_schema.str_schema(), default="disk")
        ),
        "location": core_schema.typed_dict_field(core_schema.str_schema()),
    })

    network_schema = core_schema.typed_dict_schema({
        "type": core_schema.typed_dict_field(
            core_schema.with_default_schema(core_schema.str_schema(), default="network")
        ),
        "endpoint": core_schema.typed_dict_field(core_schema.str_schema()),
    })

    union_schema = core_schema.tagged_union_schema(
        choices={"disk": disk_schema, "network": network_schema},
        discriminator="type",
    )
    return SchemaValidator(union_schema)


def build_string_multiplier_validator(factor: int = 2) -> SchemaValidator:
    """Build a SchemaValidator with chain_schema converting string to int and multiplying."""
    chain = core_schema.chain_schema([
        core_schema.str_schema(),
        core_schema.with_info_plain_validator_function(lambda v, info: int(v) * factor),
    ])
    return SchemaValidator(chain)

import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

# Add workspace root to sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import pydantic_core
from pydantic_core import (
    MultiHostUrl,
    PydanticKnownError,
    SchemaValidator,
    TzInfo,
    Url,
    ValidationError,
    to_jsonable_python,
)
from src.utils import (
    build_string_multiplier_validator,
    build_tagged_storage_validator,
    convert_to_jsonable,
    create_offset_datetime,
    parse_cluster_hosts,
    parse_endpoint,
)

# 1. Package version check
assert pydantic_core.__version__ == "2.46.4"

# 2. Url parsing, component inspection, and query extraction
url_str = "https://example.com/v1/metrics?format=json&interval=60#overview"
endpoint_info = parse_endpoint(url_str)
assert endpoint_info["scheme"] == "https"
assert endpoint_info["host"] == "example.com"
assert endpoint_info["port"] == 443
assert endpoint_info["path"] == "/v1/metrics"
assert endpoint_info["query"] == "format=json&interval=60"
assert endpoint_info["fragment"] == "overview"
assert endpoint_info["query_params"] == {"format": "json", "interval": "60"}

raw_url = Url(url_str)
assert str(raw_url) == url_str
assert raw_url.query_params() == [("format", "json"), ("interval", "60")]

# 3. MultiHostUrl connection string parsing and host enumeration
cluster_url = "postgresql://example.com:5432,example.org:5433/appdb?sslmode=require"
cluster_hosts = parse_cluster_hosts(cluster_url)
assert len(cluster_hosts) == 2
assert cluster_hosts[0] == ("example.com", 5432)
assert cluster_hosts[1] == ("example.org", 5433)

mhu = MultiHostUrl(cluster_url)
assert mhu.scheme == "postgresql"
assert mhu.path == "/appdb"
assert mhu.query == "sslmode=require"

# 4. TzInfo fixed timezone offsets with standard datetime
dt_with_offset = create_offset_datetime(2026, 9, 5, 14, 0, 0, offset_seconds=3600)
assert dt_with_offset.utcoffset() == timedelta(hours=1)
assert dt_with_offset.isoformat() == "2026-09-05T14:00:00+01:00"

tz_raw = TzInfo(7200)
dt_custom = datetime(2026, 9, 5, 15, 0, 0, tzinfo=tz_raw)
assert dt_custom.utcoffset() == timedelta(hours=2)

# 5. to_jsonable_python conversion of non-primitive Python types
dt_utc = datetime(2026, 9, 5, 12, 0, 0, tzinfo=timezone.utc)
complex_data = {
    "timestamp": dt_utc,
    "unique_tags": {"alpha", "beta"},
    "pair": (10, 20),
}
jsonable = convert_to_jsonable(complex_data)
assert jsonable["timestamp"] == "2026-09-05T12:00:00Z"
assert sorted(jsonable["unique_tags"]) == ["alpha", "beta"]
assert jsonable["pair"] == [10, 20]

direct_jsonable = to_jsonable_python({"coords": (1.5, 2.5)})
assert direct_jsonable["coords"] == [1.5, 2.5]

# 6. PydanticKnownError template and context metadata
known_err = PydanticKnownError("greater_than", {"gt": 50})
assert known_err.message_template == "Input should be greater than {gt}"
assert known_err.context == {"gt": 50}

# 7. Tagged union schema with discriminator resolution and error handling
storage_validator = build_tagged_storage_validator()

valid_disk = storage_validator.validate_python({"type": "disk", "location": "data/store"})
assert valid_disk["type"] == "disk"
assert valid_disk["location"] == "data/store"

valid_network = storage_validator.validate_python(
    {"type": "network", "endpoint": "https://example.com/stream"}
)
assert valid_network["type"] == "network"
assert valid_network["endpoint"] == "https://example.com/stream"

try:
    storage_validator.validate_python({"type": "memory", "capacity": 1024})
    raise AssertionError("Expected ValidationError for unknown discriminator tag")
except ValidationError as exc:
    assert exc.error_count() == 1
    assert exc.errors()[0]["type"] == "union_tag_invalid"

# 8. Chained schema validation and transformation pipeline
chain_validator = build_string_multiplier_validator(factor=3)
result_val = chain_validator.validate_python("14")
assert result_val == 42

print("CONTRACT PASS: pkg:pypi/pydantic-core@2.46.4 pydantic_core verified offline")