cocoindex-io/cocoindex · error · TypeError

record_type must be a record type (dataclass, NamedTuple…

Error message

record_type must be a record type (dataclass, NamedTuple, Pydantic model), got {type(record_type)}

What it means

TableTarget.from_class derives column definitions by introspecting a record type, which must be a dataclass, NamedTuple, or Pydantic model. Anything else cannot be introspected, so a TypeError is raised naming the offending type.

Solutions

Convert the type to a @dataclass, NamedTuple, or Pydantic BaseModel.
Pass the class, not an instance.
For non-record schemas, build the columns dict manually and call the constructor directly.

Example fix

// before
target = await PgTableTarget.from_class(dict, primary_key=["id"])
// after
@dataclass
class UserRow:
    id: int
    name: str
target = await PgTableTarget.from_class(UserRow, primary_key=["id"])

Defensive patterns

Strategy: type-guard

Validate before calling

import dataclasses
assert dataclasses.is_dataclass(Row) or issubclass(Row, tuple) or hasattr(Row, "model_fields")

Type guard

def is_record_class(t: object) -> bool:
    import dataclasses
    return (
        isinstance(t, type)
        and (dataclasses.is_dataclass(t)
             or issubclass(t, tuple) and hasattr(t, "_fields")
             or hasattr(t, "model_fields"))
    )

Try / catch

try:
    target = await PgTableTarget.from_class(Row, primary_key=["id"])
except TypeError as e:
    if "must be a record type" in str(e):
        raise ValueError("Pass a @dataclass/NamedTuple/BaseModel class, not an instance or dict") from e

Prevention

Define table row schemas as dataclasses and pass the class, never an instance or dict.
Avoid TypedDict here; it is not accepted as a record type.
Use typing to annotate from_class so type checkers catch wrong arguments early.

When it happens

Trigger: Calling `await PgTableTarget.from_class(dict)` or passing a TypedDict, plain class, or an instance rather than a supported record class.

Common situations: Passing a TypedDict for convenience; forgetting the @dataclass decorator; passing an instance (MyRow(...)) instead of the class.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

AI-assisted analysis of cocoindex-io/cocoindex@e84aa99b32 (2026-09-08). Data as JSON: /api/errors/7de315f3f0e6e043. Report an issue: GitHub.

Appendix: source

Thrown at python/cocoindex/connectors/postgres/_target.py:385

        primary_key: list[str],
        *,
        column_overrides: dict[str, PgType | res_schema.VectorSchemaProvider]
        | None = None,
    ) -> "TableSchema[RowT]":
        """
        Create a TableSchema from a record type (dataclass, NamedTuple, or Pydantic model).

        Python types are automatically mapped to PostgreSQL types based on asyncpg's
        type conversion.

        Args:
            record_type: A record type (dataclass, NamedTuple, or Pydantic model).
            primary_key: List of column names that form the primary key.
            column_overrides: Optional dict mapping column names to PgType or
                              VectorSchemaProvider to override the default type mapping.
        """
        if not is_record_type(record_type):
            raise TypeError(
                f"record_type must be a record type (dataclass, NamedTuple, Pydantic model), "
                f"got {type(record_type)}"
            )
        columns = await cls._columns_from_record_type(record_type, column_overrides)
        return cls(columns, primary_key, row_type=record_type)

    @staticmethod
    async def _columns_from_record_type(
        record_type: type,
        column_overrides: dict[str, PgType | res_schema.VectorSchemaProvider] | None,
    ) -> dict[str, ColumnDef]:
        """Convert a record type to a dict of column name -> ColumnDef."""
        record_info = RecordType(record_type)
        columns: dict[str, ColumnDef] = {}

        for field in record_info.fields:
            type_info = analyze_type_info(field.type_hint)

View on GitHub (pinned to e84aa99b32)