Skip to content

sql

SQLDataclass

Bases: DataclassMixin

A dataclass backed by a SQL table using the sqlalchemy ORM.

Per-field settings can be passed into the metadata argument of each dataclasses.field. See SQLDataclassFieldSettings for the full list of settings.

All dataclass fields will correspond to SQL columns unless their metadata is marked with sql=False.

Each field may also contain a "column" entry in its metadata dict. This will provide optional keyword arguments to be passed to sqlalchemy's Column constructor.

Some types are invalid for SQL columns; if such a type occurs, a TypeError will be raised.

Source code in fancy_dataclass/sql.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class SQLDataclass(DataclassMixin):
    """A dataclass backed by a SQL table using the [sqlalchemy](https://www.sqlalchemy.org) ORM.

    Per-field settings can be passed into the `metadata` argument of each `dataclasses.field`. See [`SQLDataclassFieldSettings`][fancy_dataclass.sql.SQLDataclassFieldSettings] for the full list of settings.

    All dataclass fields will correspond to SQL columns unless their metadata is marked with `sql=False`.

    Each field may also contain a `"column"` entry in its `metadata` dict. This will provide optional keyword arguments to be passed to sqlalchemy's [`Column`](https://docs.sqlalchemy.org/en/20/core/metadata.html#sqlalchemy.schema.Column) constructor.

    Some types are invalid for SQL columns; if such a type occurs, a `TypeError` will be raised."""

    __field_settings_type__ = SQLDataclassFieldSettings
    __table__: ClassVar[Table]

    @classmethod
    def get_columns(cls) -> Dict[str, Column[Any]]:
        """Gets a mapping from the class's field names to sqlalchemy `Column` objects.

        Returns:
            Dict from column names to `Column` objects"""
        cols = {}
        for fld in fields(cls):  # type: ignore[arg-type]
            settings = cls._field_settings(fld).adapt_to(SQLDataclassFieldSettings)
            nullable = False
            if not settings.sql:  # skip fields whose 'sql' setting is False
                continue
            tp: type = cast(type, fld.type)
            origin = get_origin(tp)
            if origin:  # compound type
                if origin is Union:  # use the first type of a Union (also handles Optional)
                    # column should be nullable by default if the type is optional
                    tp_args = get_args(tp)
                    nullable |= (type(None) in tp_args)
                    tp = tp_args[0]
                else:  # some other compound type
                    tp = origin
            if issubclass(tp, SQLDataclass):  # nested SQLDataclass
                cols.update(tp.get_columns())
            else:
                # TODO: making columns non-nullable seems to break things for nested SQLDataclasses
                # column_kwargs = {'nullable' : nullable}
                column_kwargs = {}
                if fld.default is not MISSING:
                    column_kwargs['default'] = fld.default
                elif fld.default_factory is not MISSING:
                    column_kwargs['default'] = fld.default_factory
                # get additional keyword arguments from 'column' section of metadata, if present
                column_kwargs.update(settings.column or {})
                cols[fld.name] = Column(fld.name, get_column_type(tp), **column_kwargs)
        return cols

get_columns() classmethod

Gets a mapping from the class's field names to sqlalchemy Column objects.

Returns:

Type Description
Dict[str, Column[Any]]

Dict from column names to Column objects
Source code in fancy_dataclass/sql.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@classmethod
def get_columns(cls) -> Dict[str, Column[Any]]:
    """Gets a mapping from the class's field names to sqlalchemy `Column` objects.

    Returns:
        Dict from column names to `Column` objects"""
    cols = {}
    for fld in fields(cls):  # type: ignore[arg-type]
        settings = cls._field_settings(fld).adapt_to(SQLDataclassFieldSettings)
        nullable = False
        if not settings.sql:  # skip fields whose 'sql' setting is False
            continue
        tp: type = cast(type, fld.type)
        origin = get_origin(tp)
        if origin:  # compound type
            if origin is Union:  # use the first type of a Union (also handles Optional)
                # column should be nullable by default if the type is optional
                tp_args = get_args(tp)
                nullable |= (type(None) in tp_args)
                tp = tp_args[0]
            else:  # some other compound type
                tp = origin
        if issubclass(tp, SQLDataclass):  # nested SQLDataclass
            cols.update(tp.get_columns())
        else:
            # TODO: making columns non-nullable seems to break things for nested SQLDataclasses
            # column_kwargs = {'nullable' : nullable}
            column_kwargs = {}
            if fld.default is not MISSING:
                column_kwargs['default'] = fld.default
            elif fld.default_factory is not MISSING:
                column_kwargs['default'] = fld.default_factory
            # get additional keyword arguments from 'column' section of metadata, if present
            column_kwargs.update(settings.column or {})
            cols[fld.name] = Column(fld.name, get_column_type(tp), **column_kwargs)
    return cols

SQLDataclassFieldSettings

Bases: FieldSettings

Settings for SQLDataclass fields.

Each field may define a metadata dict containing any of the following entries:

  • sql: if True, include this field as a table column (default True)
  • column: dict of keyword arguments passed to the Column constructor
Source code in fancy_dataclass/sql.py 
44
45
46
47
48
49
50
51
52
53
@dataclass_kw_only()
class SQLDataclassFieldSettings(FieldSettings):
    """Settings for [`SQLDataclass`][fancy_dataclass.sql.SQLDataclass] fields.

    Each field may define a `metadata` dict containing any of the following entries:

    - `sql`: if `True`, include this field as a table column (default `True`)
    - `column`: dict of keyword arguments passed to the [`Column`](https://docs.sqlalchemy.org/en/20/core/metadata.html#sqlalchemy.schema.Column) constructor"""
    sql: bool = True
    column: Optional[Dict[str, Any]] = None

get_column_type(tp)

Converts from a Python type to a corresponding sqlalchemy column type.

Parameters:

Name Type Description Default
tp type

A Python type

 required

Returns:

Type Description
type

Corresponding sqlalchemy column type
Source code in fancy_dataclass/sql.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def get_column_type(tp: type) -> type:
    """Converts from a Python type to a corresponding sqlalchemy column type.

    Args:
        tp: A Python type

    Returns:
        Corresponding sqlalchemy column type"""
    if issubclass(tp, str):
        return String
    if issubclass(tp, bool):
        return Boolean
    if issubclass(tp, int):
        return Integer
    if issubclass(tp, float):
        return Numeric
    if issubclass(tp, bytes):
        return LargeBinary
    if issubclass(tp, datetime):
        return DateTime
    return PickleType

register(reg=DEFAULT_REGISTRY, extra_cols=None)

Decorator that registers a sqlalchemy table for a SQLDataclass.

Parameters:

Name Type Description Default
reg Reg

sqlalchemy registry for mapping the class to a SQL table

 DEFAULT_REGISTRY
extra_cols Optional[Dict[str, Column[Any]]]

Additional columns (beyond the dataclass fields) to be stored in the table

 None

Returns:

Type Description
Callable[[Type[SQLDataclass]], Type[SQLDataclass]]

A decorator mapping a SQLDataclass type to a registered sqlalchemy table
Source code in fancy_dataclass/sql.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def register(reg: Reg = DEFAULT_REGISTRY, extra_cols: Optional[Dict[str, Column[Any]]] = None) -> Callable[[Type[SQLDataclass]], Type[SQLDataclass]]:
    """Decorator that registers a sqlalchemy table for a [`SQLDataclass`][fancy_dataclass.sql.SQLDataclass].

    Args:
        reg: sqlalchemy registry for mapping the class to a SQL table
        extra_cols: Additional columns (beyond the dataclass fields) to be stored in the table

    Returns:
        A decorator mapping a `SQLDataclass` type to a registered sqlalchemy table"""
    def _orm_table(cls: Type[SQLDataclass]) -> Type[SQLDataclass]:
        cls = dataclass(cls)
        cols: Dict[str, Column[Any]] = {}
        safe_dict_update(cols, cls.get_columns())
        if extra_cols:
            safe_dict_update(cols, extra_cols)
        primary_key = next((name for (name, col) in cols.items() if col.primary_key), None)
        if primary_key is None:
            if '_id' in cols:
                raise ValueError(f'no primary key found for {cls.__name__!r}')
            # add an auto-incrementing primary key with '_id' column
            cols = {'_id': Column('_id', Integer, primary_key=True, autoincrement=True), **cols}
        else:  # ensure primary key is the first column
            col = cols.pop(primary_key)
            cols = {primary_key: col, **cols}
        cls.__table__ = Table(cls.__name__, reg.metadata, *cols.values())
        return reg.mapped(cls)
    return _orm_table