GitHub-Proxy/sqlmodel-fix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✨ Add support for Pydantic v2 (while keeping support for v1 if v2 is not available), including initial work by AntonDeMeester (#722)

Browse Source 
Co-authored-by: Mohamed Farahat <farahats9@yahoo.com>
Co-authored-by: Stefan Borer <stefan.borer@gmail.com>
Co-authored-by: Peter Landry <peter.landry@gmail.com>
Co-authored-by: Anton De Meester <antondemeester+github@gmail.com>
This commit is contained in:
Sebastián Ramírez
2023-12-04 15:42:39 +01:00
committed by GitHub
parent 5b733b348d
commit fa2f178b8a
79 changed files with 2614 additions and 517 deletions
Download Patch File Download Diff File Expand all files Collapse all files

517
sqlmodel/main.py
View File

@@ -11,7 +11,6 @@ from typing import (
Callable,
ClassVar,
Dict,
ForwardRef,
List,
Mapping,
Optional,
@@ -25,13 +24,8 @@ from typing import (
overload,
)
from pydantic import BaseConfig, BaseModel
from pydantic.errors import ConfigError, DictError
from pydantic.fields import SHAPE_SINGLETON, ModelField, Undefined, UndefinedType
from pydantic import BaseModel
from pydantic.fields import FieldInfo as PydanticFieldInfo
from pydantic.main import ModelMetaclass, validate_model
from pydantic.typing import NoArgAnyCallable, resolve_annotations
from pydantic.utils import ROOT_KEY, Representation
from sqlalchemy import (
Boolean,
Column,
@@ -57,11 +51,38 @@ from sqlalchemy.orm.decl_api import DeclarativeMeta
from sqlalchemy.orm.instrumentation import is_instrumented
from sqlalchemy.sql.schema import MetaData
from sqlalchemy.sql.sqltypes import LargeBinary, Time
from typing_extensions import get_origin
from typing_extensions import Literal, deprecated, get_origin
from ._compat import ( # type: ignore[attr-defined]
IS_PYDANTIC_V2,
BaseConfig,
ModelField,
ModelMetaclass,
Representation,
SQLModelConfig,
Undefined,
UndefinedType,
_calculate_keys,
finish_init,
get_annotations,
get_config_value,
get_field_metadata,
get_model_fields,
get_relationship_to,
get_type_from_field,
is_field_noneable,
is_table_model_class,
post_init_field_info,
set_config_value,
set_fields_set,
sqlmodel_init,
sqlmodel_validate,
)
from .sql.sqltypes import GUID, AutoString
_T = TypeVar("_T")
NoArgAnyCallable = Callable[[], Any]
IncEx = Union[Set[int], Set[str], Dict[int, Any], Dict[str, Any], None]
def __dataclass_transform__(
@@ -321,7 +342,7 @@ def Field(
sa_column_kwargs=sa_column_kwargs,
**current_schema_extra,
)
field_info._validate()
post_init_field_info(field_info)
return field_info
@@ -341,7 +362,7 @@ def Relationship(
*,
back_populates: Optional[str] = None,
link_model: Optional[Any] = None,
sa_relationship: Optional[RelationshipProperty] = None, # type: ignore
sa_relationship: Optional[RelationshipProperty[Any]] = None,
) -> Any:
...
@@ -350,7 +371,7 @@ def Relationship(
*,
back_populates: Optional[str] = None,
link_model: Optional[Any] = None,
sa_relationship: Optional[RelationshipProperty] = None, # type: ignore
sa_relationship: Optional[RelationshipProperty[Any]] = None,
sa_relationship_args: Optional[Sequence[Any]] = None,
sa_relationship_kwargs: Optional[Mapping[str, Any]] = None,
) -> Any:
@@ -367,18 +388,20 @@ def Relationship(
@__dataclass_transform__(kw_only_default=True, field_descriptors=(Field, FieldInfo))
class SQLModelMetaclass(ModelMetaclass, DeclarativeMeta):
__sqlmodel_relationships__: Dict[str, RelationshipInfo]
__config__: Type[BaseConfig]
__fields__: Dict[str, ModelField]
model_config: SQLModelConfig
model_fields: Dict[str, FieldInfo]
__config__: Type[SQLModelConfig]
__fields__: Dict[str, ModelField] # type: ignore[assignment]
# Replicate SQLAlchemy
def __setattr__(cls, name: str, value: Any) -> None:
if getattr(cls.__config__, "table", False):
if is_table_model_class(cls):
DeclarativeMeta.__setattr__(cls, name, value)
else:
super().__setattr__(name, value)
def __delattr__(cls, name: str) -> None:
if getattr(cls.__config__, "table", False):
if is_table_model_class(cls):
DeclarativeMeta.__delattr__(cls, name)
else:
super().__delattr__(name)
@@ -393,9 +416,7 @@ class SQLModelMetaclass(ModelMetaclass, DeclarativeMeta):
) -> Any:
relationships: Dict[str, RelationshipInfo] = {}
dict_for_pydantic = {}
original_annotations = resolve_annotations(
class_dict.get("__annotations__", {}), class_dict.get("__module__", None)
)
original_annotations = get_annotations(class_dict)
pydantic_annotations = {}
relationship_annotations = {}
for k, v in class_dict.items():
@@ -424,10 +445,8 @@ class SQLModelMetaclass(ModelMetaclass, DeclarativeMeta):
key.startswith("__") and key.endswith("__")
) # skip dunder methods and attributes
}
pydantic_kwargs = kwargs.copy()
config_kwargs = {
key: pydantic_kwargs.pop(key)
for key in pydantic_kwargs.keys() & allowed_config_kwargs
key: kwargs[key] for key in kwargs.keys() & allowed_config_kwargs
}
new_cls = super().__new__(cls, name, bases, dict_used, **config_kwargs)
new_cls.__annotations__ = {
@@ -437,7 +456,9 @@ class SQLModelMetaclass(ModelMetaclass, DeclarativeMeta):
}
def get_config(name: str) -> Any:
config_class_value = getattr(new_cls.__config__, name, Undefined)
config_class_value = get_config_value(
model=new_cls, parameter=name, default=Undefined
)
if config_class_value is not Undefined:
return config_class_value
kwarg_value = kwargs.get(name, Undefined)
@@ -448,22 +469,27 @@ class SQLModelMetaclass(ModelMetaclass, DeclarativeMeta):
config_table = get_config("table")
if config_table is True:
# If it was passed by kwargs, ensure it's also set in config
new_cls.__config__.table = config_table
for k, v in new_cls.__fields__.items():
set_config_value(model=new_cls, parameter="table", value=config_table)
for k, v in get_model_fields(new_cls).items():
col = get_column_from_field(v)
setattr(new_cls, k, col)
# Set a config flag to tell FastAPI that this should be read with a field
# in orm_mode instead of preemptively converting it to a dict.
# This could be done by reading new_cls.__config__.table in FastAPI, but
# This could be done by reading new_cls.model_config['table'] in FastAPI, but
# that's very specific about SQLModel, so let's have another config that
# other future tools based on Pydantic can use.
new_cls.__config__.read_with_orm_mode = True
set_config_value(
model=new_cls, parameter="read_from_attributes", value=True
)
# For compatibility with older versions
# TODO: remove this in the future
set_config_value(model=new_cls, parameter="read_with_orm_mode", value=True)
config_registry = get_config("registry")
if config_registry is not Undefined:
config_registry = cast(registry, config_registry)
# If it was passed by kwargs, ensure it's also set in config
new_cls.__config__.registry = config_table
set_config_value(model=new_cls, parameter="registry", value=config_table)
setattr(new_cls, "_sa_registry", config_registry) # noqa: B010
setattr(new_cls, "metadata", config_registry.metadata) # noqa: B010
setattr(new_cls, "__abstract__", True) # noqa: B010
@@ -477,13 +503,8 @@ class SQLModelMetaclass(ModelMetaclass, DeclarativeMeta):
# this allows FastAPI cloning a SQLModel for the response_model without
# trying to create a new SQLAlchemy, for a new table, with the same name, that
# triggers an error
base_is_table = False
for base in bases:
config = getattr(base, "__config__") # noqa: B009
if config and getattr(config, "table", False):
base_is_table = True
break
if getattr(cls.__config__, "table", False) and not base_is_table:
base_is_table = any(is_table_model_class(base) for base in bases)
if is_table_model_class(cls) and not base_is_table:
for rel_name, rel_info in cls.__sqlmodel_relationships__.items():
if rel_info.sa_relationship:
# There's a SQLAlchemy relationship declared, that takes precedence
@@ -500,16 +521,9 @@ class SQLModelMetaclass(ModelMetaclass, DeclarativeMeta):
# handled well by SQLAlchemy without Mapped, so, wrap the
# annotations in Mapped here
cls.__annotations__[rel_name] = Mapped[ann] # type: ignore[valid-type]
temp_field = ModelField.infer(
name=rel_name,
value=rel_info,
annotation=ann,
class_validators=None,
config=BaseConfig,
relationship_to = get_relationship_to(
name=rel_name, rel_info=rel_info, annotation=ann
)
relationship_to = temp_field.type_
if isinstance(temp_field.type_, ForwardRef):
relationship_to = temp_field.type_.__forward_arg__
rel_kwargs: Dict[str, Any] = {}
if rel_info.back_populates:
rel_kwargs["back_populates"] = rel_info.back_populates
@@ -537,77 +551,89 @@ class SQLModelMetaclass(ModelMetaclass, DeclarativeMeta):
ModelMetaclass.__init__(cls, classname, bases, dict_, **kw)
def get_sqlalchemy_type(field: ModelField) -> Any:
sa_type = getattr(field.field_info, "sa_type", Undefined) # noqa: B009
def get_sqlalchemy_type(field: Any) -> Any:
if IS_PYDANTIC_V2:
field_info = field
else:
field_info = field.field_info
sa_type = getattr(field_info, "sa_type", Undefined) # noqa: B009
if sa_type is not Undefined:
return sa_type
if isinstance(field.type_, type) and field.shape == SHAPE_SINGLETON:
# Check enums first as an enum can also be a str, needed by Pydantic/FastAPI
if issubclass(field.type_, Enum):
return sa_Enum(field.type_)
if issubclass(field.type_, str):
if field.field_info.max_length:
return AutoString(length=field.field_info.max_length)
return AutoString
if issubclass(field.type_, float):
return Float
if issubclass(field.type_, bool):
return Boolean
if issubclass(field.type_, int):
return Integer
if issubclass(field.type_, datetime):
return DateTime
if issubclass(field.type_, date):
return Date
if issubclass(field.type_, timedelta):
return Interval
if issubclass(field.type_, time):
return Time
if issubclass(field.type_, bytes):
return LargeBinary
if issubclass(field.type_, Decimal):
return Numeric(
precision=getattr(field.type_, "max_digits", None),
scale=getattr(field.type_, "decimal_places", None),
)
if issubclass(field.type_, ipaddress.IPv4Address):
return AutoString
if issubclass(field.type_, ipaddress.IPv4Network):
return AutoString
if issubclass(field.type_, ipaddress.IPv6Address):
return AutoString
if issubclass(field.type_, ipaddress.IPv6Network):
return AutoString
if issubclass(field.type_, Path):
return AutoString
if issubclass(field.type_, uuid.UUID):
return GUID
raise ValueError(f"The field {field.name} has no matching SQLAlchemy type")
type_ = get_type_from_field(field)
metadata = get_field_metadata(field)
# Check enums first as an enum can also be a str, needed by Pydantic/FastAPI
if issubclass(type_, Enum):
return sa_Enum(type_)
if issubclass(type_, str):
max_length = getattr(metadata, "max_length", None)
if max_length:
return AutoString(length=max_length)
return AutoString
if issubclass(type_, float):
return Float
if issubclass(type_, bool):
return Boolean
if issubclass(type_, int):
return Integer
if issubclass(type_, datetime):
return DateTime
if issubclass(type_, date):
return Date
if issubclass(type_, timedelta):
return Interval
if issubclass(type_, time):
return Time
if issubclass(type_, bytes):
return LargeBinary
if issubclass(type_, Decimal):
return Numeric(
precision=getattr(metadata, "max_digits", None),
scale=getattr(metadata, "decimal_places", None),
)
if issubclass(type_, ipaddress.IPv4Address):
return AutoString
if issubclass(type_, ipaddress.IPv4Network):
return AutoString
if issubclass(type_, ipaddress.IPv6Address):
return AutoString
if issubclass(type_, ipaddress.IPv6Network):
return AutoString
if issubclass(type_, Path):
return AutoString
if issubclass(type_, uuid.UUID):
return GUID
raise ValueError(f"{type_} has no matching SQLAlchemy type")
def get_column_from_field(field: ModelField) -> Column: # type: ignore
sa_column = getattr(field.field_info, "sa_column", Undefined)
def get_column_from_field(field: Any) -> Column: # type: ignore
if IS_PYDANTIC_V2:
field_info = field
else:
field_info = field.field_info
sa_column = getattr(field_info, "sa_column", Undefined)
if isinstance(sa_column, Column):
return sa_column
sa_type = get_sqlalchemy_type(field)
primary_key = getattr(field.field_info, "primary_key", Undefined)
primary_key = getattr(field_info, "primary_key", Undefined)
if primary_key is Undefined:
primary_key = False
index = getattr(field.field_info, "index", Undefined)
index = getattr(field_info, "index", Undefined)
if index is Undefined:
index = False
nullable = not primary_key and _is_field_noneable(field)
nullable = not primary_key and is_field_noneable(field)
# Override derived nullability if the nullable property is set explicitly
# on the field
field_nullable = getattr(field.field_info, "nullable", Undefined) # noqa: B009
if field_nullable != Undefined:
field_nullable = getattr(field_info, "nullable", Undefined) # noqa: B009
if field_nullable is not Undefined:
assert not isinstance(field_nullable, UndefinedType)
nullable = field_nullable
args = []
foreign_key = getattr(field.field_info, "foreign_key", Undefined)
foreign_key = getattr(field_info, "foreign_key", Undefined)
if foreign_key is Undefined:
foreign_key = None
unique = getattr(field.field_info, "unique", Undefined)
unique = getattr(field_info, "unique", Undefined)
if unique is Undefined:
unique = False
if foreign_key:
@@ -620,16 +646,16 @@ def get_column_from_field(field: ModelField) -> Column: # type: ignore
"unique": unique,
}
sa_default = Undefined
if field.field_info.default_factory:
sa_default = field.field_info.default_factory
elif field.field_info.default is not Undefined:
sa_default = field.field_info.default
if field_info.default_factory:
sa_default = field_info.default_factory
elif field_info.default is not Undefined:
sa_default = field_info.default
if sa_default is not Undefined:
kwargs["default"] = sa_default
sa_column_args = getattr(field.field_info, "sa_column_args", Undefined)
sa_column_args = getattr(field_info, "sa_column_args", Undefined)
if sa_column_args is not Undefined:
args.extend(list(cast(Sequence[Any], sa_column_args)))
sa_column_kwargs = getattr(field.field_info, "sa_column_kwargs", Undefined)
sa_column_kwargs = getattr(field_info, "sa_column_kwargs", Undefined)
if sa_column_kwargs is not Undefined:
kwargs.update(cast(Dict[Any, Any], sa_column_kwargs))
return Column(sa_type, *args, **kwargs) # type: ignore
@@ -639,13 +665,6 @@ class_registry = weakref.WeakValueDictionary() # type: ignore
default_registry = registry()
def _value_items_is_true(v: Any) -> bool:
# Re-implement Pydantic's ValueItems.is_true() as it hasn't been released as of
# the current latest, Pydantic 1.8.2
return v is True or v is ...
_TSQLModel = TypeVar("_TSQLModel", bound="SQLModel")
@@ -653,13 +672,17 @@ class SQLModel(BaseModel, metaclass=SQLModelMetaclass, registry=default_registry
# SQLAlchemy needs to set weakref(s), Pydantic will set the other slots values
__slots__ = ("__weakref__",)
__tablename__: ClassVar[Union[str, Callable[..., str]]]
__sqlmodel_relationships__: ClassVar[Dict[str, RelationshipProperty]] # type: ignore
__sqlmodel_relationships__: ClassVar[Dict[str, RelationshipProperty[Any]]]
__name__: ClassVar[str]
metadata: ClassVar[MetaData]
__allow_unmapped__ = True # https://docs.sqlalchemy.org/en/20/changelog/migration_20.html#migration-20-step-six
class Config:
orm_mode = True
if IS_PYDANTIC_V2:
model_config = SQLModelConfig(from_attributes=True)
else:
class Config:
orm_mode = True
def __new__(cls, *args: Any, **kwargs: Any) -> Any:
new_object = super().__new__(cls)
@@ -668,31 +691,28 @@ class SQLModel(BaseModel, metaclass=SQLModelMetaclass, registry=default_registry
# Set __fields_set__ here, that would have been set when calling __init__
# in the Pydantic model so that when SQLAlchemy sets attributes that are
# added (e.g. when querying from DB) to the __fields_set__, this already exists
object.__setattr__(new_object, "__fields_set__", set())
set_fields_set(new_object, set())
return new_object
def __init__(__pydantic_self__, **data: Any) -> None:
# Uses something other than `self` the first arg to allow "self" as a
# settable attribute
values, fields_set, validation_error = validate_model(
__pydantic_self__.__class__, data
)
# Only raise errors if not a SQLModel model
if (
not getattr(__pydantic_self__.__config__, "table", False)
and validation_error
):
raise validation_error
# Do not set values as in Pydantic, pass them through setattr, so SQLAlchemy
# can handle them
# object.__setattr__(__pydantic_self__, '__dict__', values)
for key, value in values.items():
setattr(__pydantic_self__, key, value)
object.__setattr__(__pydantic_self__, "__fields_set__", fields_set)
non_pydantic_keys = data.keys() - values.keys()
for key in non_pydantic_keys:
if key in __pydantic_self__.__sqlmodel_relationships__:
setattr(__pydantic_self__, key, data[key])
# SQLAlchemy does very dark black magic and modifies the __init__ method in
# sqlalchemy.orm.instrumentation._generate_init()
# so, to make SQLAlchemy work, it's needed to explicitly call __init__ to
# trigger all the SQLAlchemy logic, it doesn't work using cls.__new__, setting
# attributes obj.__dict__, etc. The __init__ method has to be called. But
# there are cases where calling all the default logic is not ideal, e.g.
# when calling Model.model_validate(), as the validation is done outside
# of instance creation.
# At the same time, __init__ is what users would normally call, by creating
# a new instance, which should have validation and all the default logic.
# So, to be able to set up the internal SQLAlchemy logic alone without
# executing the rest, and support things like Model.model_validate(), we
# use a contextvar to know if we should execute everything.
if finish_init.get():
sqlmodel_init(self=__pydantic_self__, data=data)
def __setattr__(self, name: str, value: Any) -> None:
if name in {"_sa_instance_state"}:
@@ -700,59 +720,13 @@ class SQLModel(BaseModel, metaclass=SQLModelMetaclass, registry=default_registry
return
else:
# Set in SQLAlchemy, before Pydantic to trigger events and updates
if getattr(self.__config__, "table", False) and is_instrumented(self, name): # type: ignore
if is_table_model_class(self.__class__) and is_instrumented(self, name): # type: ignore[no-untyped-call]
set_attribute(self, name, value)
# Set in Pydantic model to trigger possible validation changes, only for
# non relationship values
if name not in self.__sqlmodel_relationships__:
super().__setattr__(name, value)
@classmethod
def from_orm(
cls: Type[_TSQLModel], obj: Any, update: Optional[Dict[str, Any]] = None
) -> _TSQLModel:
# Duplicated from Pydantic
if not cls.__config__.orm_mode:
raise ConfigError(
"You must have the config attribute orm_mode=True to use from_orm"
)
obj = {ROOT_KEY: obj} if cls.__custom_root_type__ else cls._decompose_class(obj)
# SQLModel, support update dict
if update is not None:
obj = {**obj, **update}
# End SQLModel support dict
if not getattr(cls.__config__, "table", False):
# If not table, normal Pydantic code
m: _TSQLModel = cls.__new__(cls)
else:
# If table, create the new instance normally to make SQLAlchemy create
# the _sa_instance_state attribute
m = cls()
values, fields_set, validation_error = validate_model(cls, obj)
if validation_error:
raise validation_error
# Updated to trigger SQLAlchemy internal handling
if not getattr(cls.__config__, "table", False):
object.__setattr__(m, "__dict__", values)
else:
for key, value in values.items():
setattr(m, key, value)
# Continue with standard Pydantic logic
object.__setattr__(m, "__fields_set__", fields_set)
m._init_private_attributes()
return m
@classmethod
def parse_obj(
cls: Type[_TSQLModel], obj: Any, update: Optional[Dict[str, Any]] = None
) -> _TSQLModel:
obj = cls._enforce_dict_if_root(obj)
# SQLModel, support update dict
if update is not None:
obj = {**obj, **update}
# End SQLModel support dict
return super().parse_obj(obj)
def __repr_args__(self) -> Sequence[Tuple[Optional[str], Any]]:
# Don't show SQLAlchemy private attributes
return [
@@ -761,33 +735,126 @@ class SQLModel(BaseModel, metaclass=SQLModelMetaclass, registry=default_registry
if not (isinstance(k, str) and k.startswith("_sa_"))
]
# From Pydantic, override to enforce validation with dict
@classmethod
def validate(cls: Type[_TSQLModel], value: Any) -> _TSQLModel:
if isinstance(value, cls):
return value.copy() if cls.__config__.copy_on_model_validation else value
@declared_attr # type: ignore
def __tablename__(cls) -> str:
return cls.__name__.lower()
value = cls._enforce_dict_if_root(value)
if isinstance(value, dict):
values, fields_set, validation_error = validate_model(cls, value)
if validation_error:
raise validation_error
model = cls(**value)
# Reset fields set, this would have been done in Pydantic in __init__
object.__setattr__(model, "__fields_set__", fields_set)
return model
elif cls.__config__.orm_mode:
return cls.from_orm(value)
elif cls.__custom_root_type__:
return cls.parse_obj(value)
@classmethod
def model_validate(
cls: Type[_TSQLModel],
obj: Any,
*,
strict: Union[bool, None] = None,
from_attributes: Union[bool, None] = None,
context: Union[Dict[str, Any], None] = None,
update: Union[Dict[str, Any], None] = None,
) -> _TSQLModel:
return sqlmodel_validate(
cls=cls,
obj=obj,
strict=strict,
from_attributes=from_attributes,
context=context,
update=update,
)
# TODO: remove when deprecating Pydantic v1, only for compatibility
def model_dump(
self,
*,
mode: Union[Literal["json", "python"], str] = "python",
include: IncEx = None,
exclude: IncEx = None,
by_alias: bool = False,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
round_trip: bool = False,
warnings: bool = True,
) -> Dict[str, Any]:
if IS_PYDANTIC_V2:
return super().model_dump(
mode=mode,
include=include,
exclude=exclude,
by_alias=by_alias,
exclude_unset=exclude_unset,
exclude_defaults=exclude_defaults,
exclude_none=exclude_none,
round_trip=round_trip,
warnings=warnings,
)
else:
try:
value_as_dict = dict(value)
except (TypeError, ValueError) as e:
raise DictError() from e
return cls(**value_as_dict)
return super().dict(
include=include,
exclude=exclude,
by_alias=by_alias,
exclude_unset=exclude_unset,
exclude_defaults=exclude_defaults,
exclude_none=exclude_none,
)
@deprecated(
"""
🚨 `obj.dict()` was deprecated in SQLModel 0.0.14, you should
instead use `obj.model_dump()`.
"""
)
def dict(
self,
*,
include: IncEx = None,
exclude: IncEx = None,
by_alias: bool = False,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
) -> Dict[str, Any]:
return self.model_dump(
include=include,
exclude=exclude,
by_alias=by_alias,
exclude_unset=exclude_unset,
exclude_defaults=exclude_defaults,
exclude_none=exclude_none,
)
@classmethod
@deprecated(
"""
🚨 `obj.from_orm(data)` was deprecated in SQLModel 0.0.14, you should
instead use `obj.model_validate(data)`.
"""
)
def from_orm(
cls: Type[_TSQLModel], obj: Any, update: Optional[Dict[str, Any]] = None
) -> _TSQLModel:
return cls.model_validate(obj, update=update)
@classmethod
@deprecated(
"""
🚨 `obj.parse_obj(data)` was deprecated in SQLModel 0.0.14, you should
instead use `obj.model_validate(data)`.
"""
)
def parse_obj(
cls: Type[_TSQLModel], obj: Any, update: Optional[Dict[str, Any]] = None
) -> _TSQLModel:
if not IS_PYDANTIC_V2:
obj = cls._enforce_dict_if_root(obj) # type: ignore[attr-defined] # noqa
return cls.model_validate(obj, update=update)
# From Pydantic, override to only show keys from fields, omit SQLAlchemy attributes
@deprecated(
"""
🚨 You should not access `obj._calculate_keys()` directly.
It is only useful for Pydantic v1.X, you should probably upgrade to
Pydantic v2.X.
""",
category=None,
)
def _calculate_keys(
self,
include: Optional[Mapping[Union[int, str], Any]],
@@ -795,44 +862,10 @@ class SQLModel(BaseModel, metaclass=SQLModelMetaclass, registry=default_registry
exclude_unset: bool,
update: Optional[Dict[str, Any]] = None,
) -> Optional[AbstractSet[str]]:
if include is None and exclude is None and not exclude_unset:
# Original in Pydantic:
# return None
# Updated to not return SQLAlchemy attributes
# Do not include relationships as that would easily lead to infinite
# recursion, or traversing the whole database
return self.__fields__.keys() # | self.__sqlmodel_relationships__.keys()
keys: AbstractSet[str]
if exclude_unset:
keys = self.__fields_set__.copy()
else:
# Original in Pydantic:
# keys = self.__dict__.keys()
# Updated to not return SQLAlchemy attributes
# Do not include relationships as that would easily lead to infinite
# recursion, or traversing the whole database
keys = self.__fields__.keys() # | self.__sqlmodel_relationships__.keys()
if include is not None:
keys &= include.keys()
if update:
keys -= update.keys()
if exclude:
keys -= {k for k, v in exclude.items() if _value_items_is_true(v)}
return keys
@declared_attr # type: ignore
def __tablename__(cls) -> str:
return cls.__name__.lower()
def _is_field_noneable(field: ModelField) -> bool:
if not field.required:
# Taken from [Pydantic](https://github.com/samuelcolvin/pydantic/blob/v1.8.2/pydantic/fields.py#L946-L947)
return field.allow_none and (
field.shape != SHAPE_SINGLETON or not field.sub_fields
return _calculate_keys(
self,
include=include,
exclude=exclude,
exclude_unset=exclude_unset,
update=update,
)
return False