Skip to content

Data

Validated types for data engineering: SQL identifiers and Kafka topics.

Data engineering types.

Modules:

  • kafka –

    Validated types for Apache Kafka identifiers.

  • sql –

    SQL data types.

Classes:

  • TableIdentifier –

    A SQL table identifier like schema.table or catalog.schema.table.

Attributes:

KafkaTopicName module-attribute 

KafkaTopicName = Annotated[str, AfterValidator(_validate_kafka_topic_name), WithJsonSchema({'type': 'string', 'pattern': pattern, 'description': 'A valid Apache Kafka topic name.', 'examples': ['my-topic', 'events.user.created', 'topic_v2'], 'title': 'KafkaTopicName', 'minLength': 1, 'maxLength': 249})]

A valid Apache Kafka topic name (e.g. my-topic).

SqlIdentifier module-attribute 

SqlIdentifier = Annotated[str, AfterValidator(_validate_sql_identifier), WithJsonSchema({'type': 'string', 'pattern': pattern, 'description': 'A valid unquoted SQL identifier.', 'examples': ['users', '_private_col', 'TableName'], 'title': 'SqlIdentifier', 'minLength': 1})]

A valid unquoted SQL identifier (e.g. users).

TableIdentifier

Bases: str

A SQL table identifier like schema.table or catalog.schema.table.

Methods:

__get_pydantic_core_schema__ classmethod 

__get_pydantic_core_schema__(source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema

Return the Pydantic core schema for TableIdentifier.

Source code in src/pydantypes/data/sql.py 
49
50
51
52
53
54
@classmethod
def __get_pydantic_core_schema__(
    cls, source_type: Any, handler: GetCoreSchemaHandler
) -> CoreSchema:
    """Return the Pydantic core schema for TableIdentifier."""
    return _str_type_core_schema(cls, source_type, handler)

__get_pydantic_json_schema__ classmethod 

__get_pydantic_json_schema__(_core_schema: CoreSchema, handler: GetJsonSchemaHandler) -> JsonSchemaValue

Return the JSON schema for TableIdentifier.

Source code in src/pydantypes/data/sql.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
@classmethod
def __get_pydantic_json_schema__(
    cls, _core_schema: CoreSchema, handler: GetJsonSchemaHandler
) -> JsonSchemaValue:
    """Return the JSON schema for TableIdentifier."""
    return {
        "type": "string",
        "format": "sql-table-identifier",
        "pattern": cls._pattern.pattern,
        "description": "A SQL table identifier (schema.table or catalog.schema.table)",
        "examples": ["public.users", "my_catalog.my_schema.my_table"],
        "title": "TableIdentifier",
    }

__new__ 

__new__(value: str) -> TableIdentifier

Create and validate a new TableIdentifier instance.

Source code in src/pydantypes/data/sql.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def __new__(cls, value: str) -> TableIdentifier:
    """Create and validate a new TableIdentifier instance."""
    m = cls._pattern.match(value)
    if not m:
        raise PydanticCustomError(
            "table_identifier",
            "Invalid SQL table identifier: expected"
            " 'schema.table' or 'catalog.schema.table'."
            " Got: {value}",
            {"value": value},
        )
    instance = str.__new__(cls, value)
    instance.catalog = m.group("catalog")
    instance.schema_name = m.group("schema")
    instance.table_name = m.group("table")
    return instance