sqlalchemy 2.0.51: engine.URL

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 sqlalchemy.engine.URL in pkg:pypi/sqlalchemy@2.0.51
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/sqlalchemy@2.0.51
Demonstrate these symbols/APIs:
  - sqlalchemy.engine.URL

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:9633309ae4cbe1a891f35e2f45ec12186d79c24c51a124fb30500a79fa6f80f4","contract":["sqlalchemy.engine.URL.create constructs immutable database connection URLs with driver, host, port, database, and query parameters","sqlalchemy.engine.URL parses RFC-1738 database connection strings and raises ArgumentError on malformed URLs","sqlalchemy.engine.URL.set returns new URL instances with updated attributes while preserving original object state","sqlalchemy.engine.URL supports query parameter modification and secure string rendering with password masking"],"goal":"verify sqlalchemy.engine.URL in pkg:pypi/sqlalchemy@2.0.51","kind":"HOW","packages":["pkg:pypi/sqlalchemy@2.0.51"],"schemaVersion":1,"symbols":["sqlalchemy.engine.URL"]},"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/sqlalchemy@2.0.51"],"schemaVersion":1,"subject":"pkg:pypi/sqlalchemy@2.0.51","symbols":["sqlalchemy.engine.URL"],"verifierAdapter":"python@1"}

sqlalchemy==2.0.51

"""Demonstration of database URL creation, manipulation, and rendering using sqlalchemy.engine.URL."""

from typing import Any, Dict, Optional, Sequence
from sqlalchemy.engine import URL, make_url


def create_connection_url(
    drivername: str,
    username: Optional[str] = None,
    password: Optional[str] = None,
    host: Optional[str] = None,
    port: Optional[int] = None,
    database: Optional[str] = None,
    query: Optional[Dict[str, Any]] = None,
) -> URL:
    """Construct an immutable URL instance using URL.create."""
    return URL.create(
        drivername=drivername,
        username=username,
        password=password,
        host=host,
        port=port,
        database=database,
        query=query or {},
    )


def parse_url_string(url_string: str) -> URL:
    """Parse an RFC-1738 database URL string into a URL instance."""
    return make_url(url_string)


def update_url_attributes(url: URL, **kwargs: Any) -> URL:
    """Return a new URL instance with updated attributes using URL.set."""
    return url.set(**kwargs)


def set_query_parameters(url: URL, query_params: Dict[str, Any]) -> URL:
    """Return a new URL with updated query parameters."""
    return url.update_query_dict(query_params)


def drop_query_parameters(url: URL, param_names: Sequence[str]) -> URL:
    """Return a new URL with specified query parameters removed."""
    return url.difference_update_query(param_names)


def render_url(url: URL, hide_password: bool = True) -> str:
    """Render the URL as a string with optional password masking."""
    return url.render_as_string(hide_password=hide_password)


def extract_dialect_info(url: URL) -> Dict[str, str]:
    """Extract backend and driver identifiers from a URL."""
    return {
        "backend": url.get_backend_name(),
        "driver": url.get_driver_name(),
    }

{
  "schemaVersion": 1,
  "goal": "verify sqlalchemy.engine.URL in pkg:pypi/sqlalchemy@2.0.51",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/sqlalchemy@2.0.51"
  ],
  "symbols": [
    "sqlalchemy.engine.URL"
  ]
}

"""Contract verification tests for sqlalchemy.engine.URL."""

import os
import sys

# Ensure sample root directory is in sys.path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from sqlalchemy.engine import URL
from sqlalchemy.exc import ArgumentError
import sample


def test_url_create_and_attributes() -> None:
    """Assertion 1: sqlalchemy.engine.URL.create constructs immutable database connection URLs with driver, host, port, database, and query parameters."""
    url = sample.create_connection_url(
        drivername="postgresql+psycopg2",
        username="dbuser",
        password="secret_password",
        host="localhost",
        port=5432,
        database="testdb",
        query={"sslmode": "require"},
    )

    assert isinstance(url, URL)
    assert url.drivername == "postgresql+psycopg2"
    assert url.username == "dbuser"
    assert url.password == "secret_password"
    assert url.host == "localhost"
    assert url.port == 5432
    assert url.database == "testdb"
    assert url.query["sslmode"] == "require"

    info = sample.extract_dialect_info(url)
    assert info["backend"] == "postgresql"
    assert info["driver"] == "psycopg2"


def test_url_parse_rfc1738_and_validation() -> None:
    """Assertion 2: sqlalchemy.engine.URL parses RFC-1738 database connection strings and raises ArgumentError on malformed URLs."""
    valid_url_str = "sqlite+pysqlite:///relative/path.db?timeout=10"
    parsed = sample.parse_url_string(valid_url_str)

    assert isinstance(parsed, URL)
    assert parsed.drivername == "sqlite+pysqlite"
    assert parsed.database == "relative/path.db"
    assert parsed.query["timeout"] == "10"

    info = sample.extract_dialect_info(parsed)
    assert info["backend"] == "sqlite"
    assert info["driver"] == "pysqlite"

    try:
        sample.parse_url_string("malformed url :///")
        assert False, "Expected ArgumentError when parsing malformed URL string"
    except ArgumentError as exc:
        assert isinstance(exc, ArgumentError)


def test_url_immutability_and_set() -> None:
    """Assertion 3: sqlalchemy.engine.URL.set returns new URL instances with updated attributes while preserving original object state."""
    original = sample.create_connection_url(
        drivername="postgresql+psycopg2",
        host="localhost",
        database="primary_db",
    )

    modified = sample.update_url_attributes(original, database="replica_db", port=5433)

    assert isinstance(modified, URL)
    assert modified is not original

    # Original retains initial values
    assert original.database == "primary_db"
    assert original.port is None

    # Modified contains updated values
    assert modified.database == "replica_db"
    assert modified.port == 5433
    assert modified.host == "localhost"
    assert modified.drivername == "postgresql+psycopg2"


def test_url_query_modification_and_rendering() -> None:
    """Assertion 4: sqlalchemy.engine.URL supports query parameter modification and secure string rendering with password masking."""
    url = sample.create_connection_url(
        drivername="postgresql",
        username="dbuser",
        password="super_secret_token",
        host="db.example.com",
        database="analytics",
        query={"sslmode": "require"},
    )

    # Query addition
    with_params = sample.set_query_parameters(url, {"client_encoding": "utf8"})
    assert with_params.query["sslmode"] == "require"
    assert with_params.query["client_encoding"] == "utf8"
    assert "client_encoding" not in url.query

    # Query removal
    without_params = sample.drop_query_parameters(with_params, ["sslmode"])
    assert "sslmode" not in without_params.query
    assert without_params.query["client_encoding"] == "utf8"

    # Password masking in rendering
    masked = sample.render_url(url, hide_password=True)
    assert "super_secret_token" not in masked
    assert "***" in masked
    assert str(url) == masked

    # Cleartext rendering
    unmasked = sample.render_url(url, hide_password=False)
    assert "super_secret_token" in unmasked
    assert "***" not in unmasked


def main() -> None:
    test_url_create_and_attributes()
    test_url_parse_rfc1738_and_validation()
    test_url_immutability_and_set()
    test_url_query_modification_and_rendering()
    print("All sqlalchemy.engine.URL contract assertions passed successfully.")


if __name__ == "__main__":
    main()