Module arti.partitions
None
None
View Source
from __future__ import annotations
__path__ = __import__("pkgutil").extend_path(__path__, __name__)
import abc
from datetime import date
from inspect import getattr_static
from typing import Any, ClassVar
from arti.fingerprints import Fingerprint
from arti.internal.models import Model
from arti.internal.utils import classproperty, frozendict, register
from arti.types import Collection, Date, Int8, Int16, Int32, Int64, Null, Type
class key_component(property):
pass
class PartitionKey(Model):
_abstract_ = True
_by_type_: ClassVar[dict[type[Type], type[PartitionKey]]] = {}
default_key_components: ClassVar[frozendict[str, str]]
matching_type: ClassVar[type[Type]]
@classmethod
def __init_subclass__(cls, **kwargs: Any) -> None:
super().__init_subclass__(**kwargs)
if cls._abstract_:
return
for attr in ("default_key_components", "matching_type"):
if not hasattr(cls, attr):
raise TypeError(f"{cls.__name__} must set `{attr}`")
if unknown := set(cls.default_key_components) - cls.key_components:
raise TypeError(
f"Unknown key_components in {cls.__name__}.default_key_components: {unknown}"
)
register(cls._by_type_, cls.matching_type, cls)
@classproperty
def key_components(cls) -> frozenset[str]:
return frozenset(cls.__fields__) | frozenset(
name for name in dir(cls) if isinstance(getattr_static(cls, name), key_component)
)
@classmethod
@abc.abstractmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
raise NotImplementedError(f"Unable to parse '{cls.__name__}' from: {key_components}")
@classmethod
def get_class_for(cls, type_: Type) -> type[PartitionKey]:
return cls._by_type_[type(type_)]
@classmethod
def types_from(cls, type_: Type) -> CompositeKeyTypes:
if not isinstance(type_, Collection):
return frozendict()
return frozendict(
{name: cls.get_class_for(field) for name, field in type_.partition_fields.items()}
)
# CompositeKey is the set of named PartitionKeys that uniquely identify a single partition.
CompositeKey = frozendict[str, PartitionKey]
CompositeKeyTypes = frozendict[str, type[PartitionKey]]
NotPartitioned = CompositeKey()
class DateKey(PartitionKey):
default_key_components: ClassVar[frozendict[str, str]] = frozendict(Y="Y", m="m:02", d="d:02")
matching_type = Date
key: date
@key_component
def Y(self) -> int:
return self.key.year
@key_component
def m(self) -> int:
return self.key.month
@key_component
def d(self) -> int:
return self.key.day
@key_component
def iso(self) -> str:
return self.key.isoformat()
@classmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
names = set(key_components)
if names == {"key"}:
return cls(key=date.fromisoformat(key_components["key"]))
if names == {"iso"}:
return cls(key=date.fromisoformat(key_components["iso"]))
if names == {"Y", "m", "d"}:
return cls(key=date(*[int(key_components[k]) for k in ("Y", "m", "d")]))
return super().from_key_components(**key_components)
class _IntKey(PartitionKey):
_abstract_ = True
default_key_components: ClassVar[frozendict[str, str]] = frozendict(key="key")
key: int
@key_component
def hex(self) -> str:
return hex(self.key)
@classmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
names = set(key_components)
if names == {"key"}:
return cls(key=int(key_components["key"]))
if names == {"hex"}:
return cls(key=int(key_components["hex"], base=16))
return super().from_key_components(**key_components)
class Int8Key(_IntKey):
matching_type = Int8
class Int16Key(_IntKey):
matching_type = Int16
class Int32Key(_IntKey):
matching_type = Int32
class Int64Key(_IntKey):
matching_type = Int64
class NullKey(PartitionKey):
default_key_components: ClassVar[frozendict[str, str]] = frozendict(key="key")
matching_type = Null
key: None = None
@classmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
if set(key_components) == {"key"}:
if key_components["key"] != "None":
raise ValueError(f"'{cls.__name__}' can only be used with 'None'!")
return cls()
return super().from_key_components(**key_components)
InputFingerprints = frozendict[CompositeKey, Fingerprint]
Variables
CompositeKey
CompositeKeyTypes
InputFingerprints
NotPartitioned
Classes
DateKey
class DateKey(
__pydantic_self__,
**data: Any
)
View Source
class DateKey(PartitionKey):
default_key_components: ClassVar[frozendict[str, str]] = frozendict(Y="Y", m="m:02", d="d:02")
matching_type = Date
key: date
@key_component
def Y(self) -> int:
return self.key.year
@key_component
def m(self) -> int:
return self.key.month
@key_component
def d(self) -> int:
return self.key.day
@key_component
def iso(self) -> str:
return self.key.isoformat()
@classmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
names = set(key_components)
if names == {"key"}:
return cls(key=date.fromisoformat(key_components["key"]))
if names == {"iso"}:
return cls(key=date.fromisoformat(key_components["iso"]))
if names == {"Y", "m", "d"}:
return cls(key=date(*[int(key_components[k]) for k in ("Y", "m", "d")]))
return super().from_key_components(**key_components)
Ancestors (in MRO)
- arti.partitions.PartitionKey
- arti.internal.models.Model
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
Config
default_key_components
key_components
matching_type
Static methods
construct
def construct(
_fields_set: Optional[ForwardRef('SetStr')] = None,
**values: Any
) -> 'Model'
Creates a new model setting dict and fields_set from trusted or pre-validated data.
Default values are respected, but no other validation is performed.
Behaves as if Config.extra = 'allow' was set since it adds all passed values
from_key_components
def from_key_components(
**key_components: 'str'
) -> 'PartitionKey'
View Source
@classmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
names = set(key_components)
if names == {"key"}:
return cls(key=date.fromisoformat(key_components["key"]))
if names == {"iso"}:
return cls(key=date.fromisoformat(key_components["iso"]))
if names == {"Y", "m", "d"}:
return cls(key=date(*[int(key_components[k]) for k in ("Y", "m", "d")]))
return super().from_key_components(**key_components)
from_orm
def from_orm(
obj: Any
) -> 'Model'
get_class_for
def get_class_for(
type_: 'Type'
) -> 'type[PartitionKey]'
View Source
@classmethod
def get_class_for(cls, type_: Type) -> type[PartitionKey]:
return cls._by_type_[type(type_)]
parse_file
def parse_file(
path: Union[str, pathlib.Path],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
parse_obj
def parse_obj(
obj: Any
) -> 'Model'
parse_raw
def parse_raw(
b: Union[str, bytes],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
schema
def schema(
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}'
) -> 'DictStrAny'
schema_json
def schema_json(
*,
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}',
**dumps_kwargs: Any
) -> 'unicode'
types_from
def types_from(
type_: 'Type'
) -> 'CompositeKeyTypes'
View Source
@classmethod
def types_from(cls, type_: Type) -> CompositeKeyTypes:
if not isinstance(type_, Collection):
return frozendict()
return frozendict(
{name: cls.get_class_for(field) for name, field in type_.partition_fields.items()}
)
update_forward_refs
def update_forward_refs(
**localns: Any
) -> None
Try to update ForwardRefs on fields based on this Model, globalns and localns.
validate
def validate(
value: Any
) -> 'Model'
Instance variables
Y
d
fingerprint
iso
m
Methods
copy
def copy(
self,
*,
deep: 'bool' = False,
validate: 'bool' = True,
**kwargs: 'Any'
) -> 'Self'
Duplicate a model, optionally choose which fields to include, exclude and change.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| include | None | fields to include in new model | None |
| exclude | None | fields to exclude from new model, as with values this takes precedence over include | None |
| update | None | values to change/add in the new model. Note: the data is not validated before creating | |
| the new model: you should trust this data | None | ||
| deep | None | set to True to make a deep copy of the model |
None |
Returns:
| Type | Description |
|---|---|
| None | new model instance |
View Source
def copy(self, *, deep: bool = False, validate: bool = True, **kwargs: Any) -> Self:
copy = super().copy(deep=deep, **kwargs)
if validate:
# NOTE: We set exclude_unset=False so that all existing defaulted fields are reused (as
# is normal `.copy` behavior).
#
# To reduce `repr` noise, we'll reset .__fields_set__ to those of the pre-validation copy
# (which includes those originally set + updated).
fields_set = copy.__fields_set__
copy = copy.validate(
dict(copy._iter(to_dict=False, by_alias=False, exclude_unset=False))
)
# Use object.__setattr__ to bypass frozen model assignment errors
object.__setattr__(copy, "__fields_set__", set(fields_set))
# Copy over the private attributes, which are missing after validation (since we're only
# passing the fields).
for name in self.__private_attributes__:
if (value := getattr(self, name, Undefined)) is not Undefined:
if deep:
value = deepcopy(value)
object.__setattr__(copy, name, value)
return copy
dict
def dict(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False
) -> 'DictStrAny'
Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.
json
def json(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
encoder: Optional[Callable[[Any], Any]] = None,
models_as_dict: bool = True,
**dumps_kwargs: Any
) -> 'unicode'
Generate a JSON representation of the model, include and exclude arguments as per dict().
encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().
Int16Key
class Int16Key(
__pydantic_self__,
**data: Any
)
View Source
class Int16Key(_IntKey):
matching_type = Int16
Ancestors (in MRO)
- arti.partitions._IntKey
- arti.partitions.PartitionKey
- arti.internal.models.Model
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
Config
default_key_components
key_components
matching_type
Static methods
construct
def construct(
_fields_set: Optional[ForwardRef('SetStr')] = None,
**values: Any
) -> 'Model'
Creates a new model setting dict and fields_set from trusted or pre-validated data.
Default values are respected, but no other validation is performed.
Behaves as if Config.extra = 'allow' was set since it adds all passed values
from_key_components
def from_key_components(
**key_components: 'str'
) -> 'PartitionKey'
View Source
@classmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
names = set(key_components)
if names == {"key"}:
return cls(key=int(key_components["key"]))
if names == {"hex"}:
return cls(key=int(key_components["hex"], base=16))
return super().from_key_components(**key_components)
from_orm
def from_orm(
obj: Any
) -> 'Model'
get_class_for
def get_class_for(
type_: 'Type'
) -> 'type[PartitionKey]'
View Source
@classmethod
def get_class_for(cls, type_: Type) -> type[PartitionKey]:
return cls._by_type_[type(type_)]
parse_file
def parse_file(
path: Union[str, pathlib.Path],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
parse_obj
def parse_obj(
obj: Any
) -> 'Model'
parse_raw
def parse_raw(
b: Union[str, bytes],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
schema
def schema(
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}'
) -> 'DictStrAny'
schema_json
def schema_json(
*,
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}',
**dumps_kwargs: Any
) -> 'unicode'
types_from
def types_from(
type_: 'Type'
) -> 'CompositeKeyTypes'
View Source
@classmethod
def types_from(cls, type_: Type) -> CompositeKeyTypes:
if not isinstance(type_, Collection):
return frozendict()
return frozendict(
{name: cls.get_class_for(field) for name, field in type_.partition_fields.items()}
)
update_forward_refs
def update_forward_refs(
**localns: Any
) -> None
Try to update ForwardRefs on fields based on this Model, globalns and localns.
validate
def validate(
value: Any
) -> 'Model'
Instance variables
fingerprint
hex
Methods
copy
def copy(
self,
*,
deep: 'bool' = False,
validate: 'bool' = True,
**kwargs: 'Any'
) -> 'Self'
Duplicate a model, optionally choose which fields to include, exclude and change.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| include | None | fields to include in new model | None |
| exclude | None | fields to exclude from new model, as with values this takes precedence over include | None |
| update | None | values to change/add in the new model. Note: the data is not validated before creating | |
| the new model: you should trust this data | None | ||
| deep | None | set to True to make a deep copy of the model |
None |
Returns:
| Type | Description |
|---|---|
| None | new model instance |
View Source
def copy(self, *, deep: bool = False, validate: bool = True, **kwargs: Any) -> Self:
copy = super().copy(deep=deep, **kwargs)
if validate:
# NOTE: We set exclude_unset=False so that all existing defaulted fields are reused (as
# is normal `.copy` behavior).
#
# To reduce `repr` noise, we'll reset .__fields_set__ to those of the pre-validation copy
# (which includes those originally set + updated).
fields_set = copy.__fields_set__
copy = copy.validate(
dict(copy._iter(to_dict=False, by_alias=False, exclude_unset=False))
)
# Use object.__setattr__ to bypass frozen model assignment errors
object.__setattr__(copy, "__fields_set__", set(fields_set))
# Copy over the private attributes, which are missing after validation (since we're only
# passing the fields).
for name in self.__private_attributes__:
if (value := getattr(self, name, Undefined)) is not Undefined:
if deep:
value = deepcopy(value)
object.__setattr__(copy, name, value)
return copy
dict
def dict(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False
) -> 'DictStrAny'
Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.
json
def json(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
encoder: Optional[Callable[[Any], Any]] = None,
models_as_dict: bool = True,
**dumps_kwargs: Any
) -> 'unicode'
Generate a JSON representation of the model, include and exclude arguments as per dict().
encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().
Int32Key
class Int32Key(
__pydantic_self__,
**data: Any
)
View Source
class Int32Key(_IntKey):
matching_type = Int32
Ancestors (in MRO)
- arti.partitions._IntKey
- arti.partitions.PartitionKey
- arti.internal.models.Model
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
Config
default_key_components
key_components
matching_type
Static methods
construct
def construct(
_fields_set: Optional[ForwardRef('SetStr')] = None,
**values: Any
) -> 'Model'
Creates a new model setting dict and fields_set from trusted or pre-validated data.
Default values are respected, but no other validation is performed.
Behaves as if Config.extra = 'allow' was set since it adds all passed values
from_key_components
def from_key_components(
**key_components: 'str'
) -> 'PartitionKey'
View Source
@classmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
names = set(key_components)
if names == {"key"}:
return cls(key=int(key_components["key"]))
if names == {"hex"}:
return cls(key=int(key_components["hex"], base=16))
return super().from_key_components(**key_components)
from_orm
def from_orm(
obj: Any
) -> 'Model'
get_class_for
def get_class_for(
type_: 'Type'
) -> 'type[PartitionKey]'
View Source
@classmethod
def get_class_for(cls, type_: Type) -> type[PartitionKey]:
return cls._by_type_[type(type_)]
parse_file
def parse_file(
path: Union[str, pathlib.Path],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
parse_obj
def parse_obj(
obj: Any
) -> 'Model'
parse_raw
def parse_raw(
b: Union[str, bytes],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
schema
def schema(
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}'
) -> 'DictStrAny'
schema_json
def schema_json(
*,
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}',
**dumps_kwargs: Any
) -> 'unicode'
types_from
def types_from(
type_: 'Type'
) -> 'CompositeKeyTypes'
View Source
@classmethod
def types_from(cls, type_: Type) -> CompositeKeyTypes:
if not isinstance(type_, Collection):
return frozendict()
return frozendict(
{name: cls.get_class_for(field) for name, field in type_.partition_fields.items()}
)
update_forward_refs
def update_forward_refs(
**localns: Any
) -> None
Try to update ForwardRefs on fields based on this Model, globalns and localns.
validate
def validate(
value: Any
) -> 'Model'
Instance variables
fingerprint
hex
Methods
copy
def copy(
self,
*,
deep: 'bool' = False,
validate: 'bool' = True,
**kwargs: 'Any'
) -> 'Self'
Duplicate a model, optionally choose which fields to include, exclude and change.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| include | None | fields to include in new model | None |
| exclude | None | fields to exclude from new model, as with values this takes precedence over include | None |
| update | None | values to change/add in the new model. Note: the data is not validated before creating | |
| the new model: you should trust this data | None | ||
| deep | None | set to True to make a deep copy of the model |
None |
Returns:
| Type | Description |
|---|---|
| None | new model instance |
View Source
def copy(self, *, deep: bool = False, validate: bool = True, **kwargs: Any) -> Self:
copy = super().copy(deep=deep, **kwargs)
if validate:
# NOTE: We set exclude_unset=False so that all existing defaulted fields are reused (as
# is normal `.copy` behavior).
#
# To reduce `repr` noise, we'll reset .__fields_set__ to those of the pre-validation copy
# (which includes those originally set + updated).
fields_set = copy.__fields_set__
copy = copy.validate(
dict(copy._iter(to_dict=False, by_alias=False, exclude_unset=False))
)
# Use object.__setattr__ to bypass frozen model assignment errors
object.__setattr__(copy, "__fields_set__", set(fields_set))
# Copy over the private attributes, which are missing after validation (since we're only
# passing the fields).
for name in self.__private_attributes__:
if (value := getattr(self, name, Undefined)) is not Undefined:
if deep:
value = deepcopy(value)
object.__setattr__(copy, name, value)
return copy
dict
def dict(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False
) -> 'DictStrAny'
Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.
json
def json(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
encoder: Optional[Callable[[Any], Any]] = None,
models_as_dict: bool = True,
**dumps_kwargs: Any
) -> 'unicode'
Generate a JSON representation of the model, include and exclude arguments as per dict().
encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().
Int64Key
class Int64Key(
__pydantic_self__,
**data: Any
)
View Source
class Int64Key(_IntKey):
matching_type = Int64
Ancestors (in MRO)
- arti.partitions._IntKey
- arti.partitions.PartitionKey
- arti.internal.models.Model
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
Config
default_key_components
key_components
matching_type
Static methods
construct
def construct(
_fields_set: Optional[ForwardRef('SetStr')] = None,
**values: Any
) -> 'Model'
Creates a new model setting dict and fields_set from trusted or pre-validated data.
Default values are respected, but no other validation is performed.
Behaves as if Config.extra = 'allow' was set since it adds all passed values
from_key_components
def from_key_components(
**key_components: 'str'
) -> 'PartitionKey'
View Source
@classmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
names = set(key_components)
if names == {"key"}:
return cls(key=int(key_components["key"]))
if names == {"hex"}:
return cls(key=int(key_components["hex"], base=16))
return super().from_key_components(**key_components)
from_orm
def from_orm(
obj: Any
) -> 'Model'
get_class_for
def get_class_for(
type_: 'Type'
) -> 'type[PartitionKey]'
View Source
@classmethod
def get_class_for(cls, type_: Type) -> type[PartitionKey]:
return cls._by_type_[type(type_)]
parse_file
def parse_file(
path: Union[str, pathlib.Path],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
parse_obj
def parse_obj(
obj: Any
) -> 'Model'
parse_raw
def parse_raw(
b: Union[str, bytes],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
schema
def schema(
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}'
) -> 'DictStrAny'
schema_json
def schema_json(
*,
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}',
**dumps_kwargs: Any
) -> 'unicode'
types_from
def types_from(
type_: 'Type'
) -> 'CompositeKeyTypes'
View Source
@classmethod
def types_from(cls, type_: Type) -> CompositeKeyTypes:
if not isinstance(type_, Collection):
return frozendict()
return frozendict(
{name: cls.get_class_for(field) for name, field in type_.partition_fields.items()}
)
update_forward_refs
def update_forward_refs(
**localns: Any
) -> None
Try to update ForwardRefs on fields based on this Model, globalns and localns.
validate
def validate(
value: Any
) -> 'Model'
Instance variables
fingerprint
hex
Methods
copy
def copy(
self,
*,
deep: 'bool' = False,
validate: 'bool' = True,
**kwargs: 'Any'
) -> 'Self'
Duplicate a model, optionally choose which fields to include, exclude and change.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| include | None | fields to include in new model | None |
| exclude | None | fields to exclude from new model, as with values this takes precedence over include | None |
| update | None | values to change/add in the new model. Note: the data is not validated before creating | |
| the new model: you should trust this data | None | ||
| deep | None | set to True to make a deep copy of the model |
None |
Returns:
| Type | Description |
|---|---|
| None | new model instance |
View Source
def copy(self, *, deep: bool = False, validate: bool = True, **kwargs: Any) -> Self:
copy = super().copy(deep=deep, **kwargs)
if validate:
# NOTE: We set exclude_unset=False so that all existing defaulted fields are reused (as
# is normal `.copy` behavior).
#
# To reduce `repr` noise, we'll reset .__fields_set__ to those of the pre-validation copy
# (which includes those originally set + updated).
fields_set = copy.__fields_set__
copy = copy.validate(
dict(copy._iter(to_dict=False, by_alias=False, exclude_unset=False))
)
# Use object.__setattr__ to bypass frozen model assignment errors
object.__setattr__(copy, "__fields_set__", set(fields_set))
# Copy over the private attributes, which are missing after validation (since we're only
# passing the fields).
for name in self.__private_attributes__:
if (value := getattr(self, name, Undefined)) is not Undefined:
if deep:
value = deepcopy(value)
object.__setattr__(copy, name, value)
return copy
dict
def dict(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False
) -> 'DictStrAny'
Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.
json
def json(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
encoder: Optional[Callable[[Any], Any]] = None,
models_as_dict: bool = True,
**dumps_kwargs: Any
) -> 'unicode'
Generate a JSON representation of the model, include and exclude arguments as per dict().
encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().
Int8Key
class Int8Key(
__pydantic_self__,
**data: Any
)
View Source
class Int8Key(_IntKey):
matching_type = Int8
Ancestors (in MRO)
- arti.partitions._IntKey
- arti.partitions.PartitionKey
- arti.internal.models.Model
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
Config
default_key_components
key_components
matching_type
Static methods
construct
def construct(
_fields_set: Optional[ForwardRef('SetStr')] = None,
**values: Any
) -> 'Model'
Creates a new model setting dict and fields_set from trusted or pre-validated data.
Default values are respected, but no other validation is performed.
Behaves as if Config.extra = 'allow' was set since it adds all passed values
from_key_components
def from_key_components(
**key_components: 'str'
) -> 'PartitionKey'
View Source
@classmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
names = set(key_components)
if names == {"key"}:
return cls(key=int(key_components["key"]))
if names == {"hex"}:
return cls(key=int(key_components["hex"], base=16))
return super().from_key_components(**key_components)
from_orm
def from_orm(
obj: Any
) -> 'Model'
get_class_for
def get_class_for(
type_: 'Type'
) -> 'type[PartitionKey]'
View Source
@classmethod
def get_class_for(cls, type_: Type) -> type[PartitionKey]:
return cls._by_type_[type(type_)]
parse_file
def parse_file(
path: Union[str, pathlib.Path],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
parse_obj
def parse_obj(
obj: Any
) -> 'Model'
parse_raw
def parse_raw(
b: Union[str, bytes],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
schema
def schema(
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}'
) -> 'DictStrAny'
schema_json
def schema_json(
*,
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}',
**dumps_kwargs: Any
) -> 'unicode'
types_from
def types_from(
type_: 'Type'
) -> 'CompositeKeyTypes'
View Source
@classmethod
def types_from(cls, type_: Type) -> CompositeKeyTypes:
if not isinstance(type_, Collection):
return frozendict()
return frozendict(
{name: cls.get_class_for(field) for name, field in type_.partition_fields.items()}
)
update_forward_refs
def update_forward_refs(
**localns: Any
) -> None
Try to update ForwardRefs on fields based on this Model, globalns and localns.
validate
def validate(
value: Any
) -> 'Model'
Instance variables
fingerprint
hex
Methods
copy
def copy(
self,
*,
deep: 'bool' = False,
validate: 'bool' = True,
**kwargs: 'Any'
) -> 'Self'
Duplicate a model, optionally choose which fields to include, exclude and change.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| include | None | fields to include in new model | None |
| exclude | None | fields to exclude from new model, as with values this takes precedence over include | None |
| update | None | values to change/add in the new model. Note: the data is not validated before creating | |
| the new model: you should trust this data | None | ||
| deep | None | set to True to make a deep copy of the model |
None |
Returns:
| Type | Description |
|---|---|
| None | new model instance |
View Source
def copy(self, *, deep: bool = False, validate: bool = True, **kwargs: Any) -> Self:
copy = super().copy(deep=deep, **kwargs)
if validate:
# NOTE: We set exclude_unset=False so that all existing defaulted fields are reused (as
# is normal `.copy` behavior).
#
# To reduce `repr` noise, we'll reset .__fields_set__ to those of the pre-validation copy
# (which includes those originally set + updated).
fields_set = copy.__fields_set__
copy = copy.validate(
dict(copy._iter(to_dict=False, by_alias=False, exclude_unset=False))
)
# Use object.__setattr__ to bypass frozen model assignment errors
object.__setattr__(copy, "__fields_set__", set(fields_set))
# Copy over the private attributes, which are missing after validation (since we're only
# passing the fields).
for name in self.__private_attributes__:
if (value := getattr(self, name, Undefined)) is not Undefined:
if deep:
value = deepcopy(value)
object.__setattr__(copy, name, value)
return copy
dict
def dict(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False
) -> 'DictStrAny'
Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.
json
def json(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
encoder: Optional[Callable[[Any], Any]] = None,
models_as_dict: bool = True,
**dumps_kwargs: Any
) -> 'unicode'
Generate a JSON representation of the model, include and exclude arguments as per dict().
encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().
NullKey
class NullKey(
__pydantic_self__,
**data: Any
)
View Source
class NullKey(PartitionKey):
default_key_components: ClassVar[frozendict[str, str]] = frozendict(key="key")
matching_type = Null
key: None = None
@classmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
if set(key_components) == {"key"}:
if key_components["key"] != "None":
raise ValueError(f"'{cls.__name__}' can only be used with 'None'!")
return cls()
return super().from_key_components(**key_components)
Ancestors (in MRO)
- arti.partitions.PartitionKey
- arti.internal.models.Model
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
Config
default_key_components
key_components
matching_type
Static methods
construct
def construct(
_fields_set: Optional[ForwardRef('SetStr')] = None,
**values: Any
) -> 'Model'
Creates a new model setting dict and fields_set from trusted or pre-validated data.
Default values are respected, but no other validation is performed.
Behaves as if Config.extra = 'allow' was set since it adds all passed values
from_key_components
def from_key_components(
**key_components: 'str'
) -> 'PartitionKey'
View Source
@classmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
if set(key_components) == {"key"}:
if key_components["key"] != "None":
raise ValueError(f"'{cls.__name__}' can only be used with 'None'!")
return cls()
return super().from_key_components(**key_components)
from_orm
def from_orm(
obj: Any
) -> 'Model'
get_class_for
def get_class_for(
type_: 'Type'
) -> 'type[PartitionKey]'
View Source
@classmethod
def get_class_for(cls, type_: Type) -> type[PartitionKey]:
return cls._by_type_[type(type_)]
parse_file
def parse_file(
path: Union[str, pathlib.Path],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
parse_obj
def parse_obj(
obj: Any
) -> 'Model'
parse_raw
def parse_raw(
b: Union[str, bytes],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
schema
def schema(
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}'
) -> 'DictStrAny'
schema_json
def schema_json(
*,
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}',
**dumps_kwargs: Any
) -> 'unicode'
types_from
def types_from(
type_: 'Type'
) -> 'CompositeKeyTypes'
View Source
@classmethod
def types_from(cls, type_: Type) -> CompositeKeyTypes:
if not isinstance(type_, Collection):
return frozendict()
return frozendict(
{name: cls.get_class_for(field) for name, field in type_.partition_fields.items()}
)
update_forward_refs
def update_forward_refs(
**localns: Any
) -> None
Try to update ForwardRefs on fields based on this Model, globalns and localns.
validate
def validate(
value: Any
) -> 'Model'
Instance variables
fingerprint
Methods
copy
def copy(
self,
*,
deep: 'bool' = False,
validate: 'bool' = True,
**kwargs: 'Any'
) -> 'Self'
Duplicate a model, optionally choose which fields to include, exclude and change.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| include | None | fields to include in new model | None |
| exclude | None | fields to exclude from new model, as with values this takes precedence over include | None |
| update | None | values to change/add in the new model. Note: the data is not validated before creating | |
| the new model: you should trust this data | None | ||
| deep | None | set to True to make a deep copy of the model |
None |
Returns:
| Type | Description |
|---|---|
| None | new model instance |
View Source
def copy(self, *, deep: bool = False, validate: bool = True, **kwargs: Any) -> Self:
copy = super().copy(deep=deep, **kwargs)
if validate:
# NOTE: We set exclude_unset=False so that all existing defaulted fields are reused (as
# is normal `.copy` behavior).
#
# To reduce `repr` noise, we'll reset .__fields_set__ to those of the pre-validation copy
# (which includes those originally set + updated).
fields_set = copy.__fields_set__
copy = copy.validate(
dict(copy._iter(to_dict=False, by_alias=False, exclude_unset=False))
)
# Use object.__setattr__ to bypass frozen model assignment errors
object.__setattr__(copy, "__fields_set__", set(fields_set))
# Copy over the private attributes, which are missing after validation (since we're only
# passing the fields).
for name in self.__private_attributes__:
if (value := getattr(self, name, Undefined)) is not Undefined:
if deep:
value = deepcopy(value)
object.__setattr__(copy, name, value)
return copy
dict
def dict(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False
) -> 'DictStrAny'
Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.
json
def json(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
encoder: Optional[Callable[[Any], Any]] = None,
models_as_dict: bool = True,
**dumps_kwargs: Any
) -> 'unicode'
Generate a JSON representation of the model, include and exclude arguments as per dict().
encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().
PartitionKey
class PartitionKey(
__pydantic_self__,
**data: Any
)
View Source
class PartitionKey(Model):
_abstract_ = True
_by_type_: ClassVar[dict[type[Type], type[PartitionKey]]] = {}
default_key_components: ClassVar[frozendict[str, str]]
matching_type: ClassVar[type[Type]]
@classmethod
def __init_subclass__(cls, **kwargs: Any) -> None:
super().__init_subclass__(**kwargs)
if cls._abstract_:
return
for attr in ("default_key_components", "matching_type"):
if not hasattr(cls, attr):
raise TypeError(f"{cls.__name__} must set `{attr}`")
if unknown := set(cls.default_key_components) - cls.key_components:
raise TypeError(
f"Unknown key_components in {cls.__name__}.default_key_components: {unknown}"
)
register(cls._by_type_, cls.matching_type, cls)
@classproperty
def key_components(cls) -> frozenset[str]:
return frozenset(cls.__fields__) | frozenset(
name for name in dir(cls) if isinstance(getattr_static(cls, name), key_component)
)
@classmethod
@abc.abstractmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
raise NotImplementedError(f"Unable to parse '{cls.__name__}' from: {key_components}")
@classmethod
def get_class_for(cls, type_: Type) -> type[PartitionKey]:
return cls._by_type_[type(type_)]
@classmethod
def types_from(cls, type_: Type) -> CompositeKeyTypes:
if not isinstance(type_, Collection):
return frozendict()
return frozendict(
{name: cls.get_class_for(field) for name, field in type_.partition_fields.items()}
)
Ancestors (in MRO)
- arti.internal.models.Model
- pydantic.main.BaseModel
- pydantic.utils.Representation
Descendants
- arti.partitions.DateKey
- arti.partitions._IntKey
- arti.partitions.NullKey
Class variables
Config
key_components
Static methods
construct
def construct(
_fields_set: Optional[ForwardRef('SetStr')] = None,
**values: Any
) -> 'Model'
Creates a new model setting dict and fields_set from trusted or pre-validated data.
Default values are respected, but no other validation is performed.
Behaves as if Config.extra = 'allow' was set since it adds all passed values
from_key_components
def from_key_components(
**key_components: 'str'
) -> 'PartitionKey'
View Source
@classmethod
@abc.abstractmethod
def from_key_components(cls, **key_components: str) -> PartitionKey:
raise NotImplementedError(f"Unable to parse '{cls.__name__}' from: {key_components}")
from_orm
def from_orm(
obj: Any
) -> 'Model'
get_class_for
def get_class_for(
type_: 'Type'
) -> 'type[PartitionKey]'
View Source
@classmethod
def get_class_for(cls, type_: Type) -> type[PartitionKey]:
return cls._by_type_[type(type_)]
parse_file
def parse_file(
path: Union[str, pathlib.Path],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
parse_obj
def parse_obj(
obj: Any
) -> 'Model'
parse_raw
def parse_raw(
b: Union[str, bytes],
*,
content_type: 'unicode' = None,
encoding: 'unicode' = 'utf8',
proto: pydantic.parse.Protocol = None,
allow_pickle: bool = False
) -> 'Model'
schema
def schema(
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}'
) -> 'DictStrAny'
schema_json
def schema_json(
*,
by_alias: bool = True,
ref_template: 'unicode' = '#/definitions/{model}',
**dumps_kwargs: Any
) -> 'unicode'
types_from
def types_from(
type_: 'Type'
) -> 'CompositeKeyTypes'
View Source
@classmethod
def types_from(cls, type_: Type) -> CompositeKeyTypes:
if not isinstance(type_, Collection):
return frozendict()
return frozendict(
{name: cls.get_class_for(field) for name, field in type_.partition_fields.items()}
)
update_forward_refs
def update_forward_refs(
**localns: Any
) -> None
Try to update ForwardRefs on fields based on this Model, globalns and localns.
validate
def validate(
value: Any
) -> 'Model'
Instance variables
fingerprint
Methods
copy
def copy(
self,
*,
deep: 'bool' = False,
validate: 'bool' = True,
**kwargs: 'Any'
) -> 'Self'
Duplicate a model, optionally choose which fields to include, exclude and change.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| include | None | fields to include in new model | None |
| exclude | None | fields to exclude from new model, as with values this takes precedence over include | None |
| update | None | values to change/add in the new model. Note: the data is not validated before creating | |
| the new model: you should trust this data | None | ||
| deep | None | set to True to make a deep copy of the model |
None |
Returns:
| Type | Description |
|---|---|
| None | new model instance |
View Source
def copy(self, *, deep: bool = False, validate: bool = True, **kwargs: Any) -> Self:
copy = super().copy(deep=deep, **kwargs)
if validate:
# NOTE: We set exclude_unset=False so that all existing defaulted fields are reused (as
# is normal `.copy` behavior).
#
# To reduce `repr` noise, we'll reset .__fields_set__ to those of the pre-validation copy
# (which includes those originally set + updated).
fields_set = copy.__fields_set__
copy = copy.validate(
dict(copy._iter(to_dict=False, by_alias=False, exclude_unset=False))
)
# Use object.__setattr__ to bypass frozen model assignment errors
object.__setattr__(copy, "__fields_set__", set(fields_set))
# Copy over the private attributes, which are missing after validation (since we're only
# passing the fields).
for name in self.__private_attributes__:
if (value := getattr(self, name, Undefined)) is not Undefined:
if deep:
value = deepcopy(value)
object.__setattr__(copy, name, value)
return copy
dict
def dict(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False
) -> 'DictStrAny'
Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.
json
def json(
self,
*,
include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
by_alias: bool = False,
skip_defaults: Optional[bool] = None,
exclude_unset: bool = False,
exclude_defaults: bool = False,
exclude_none: bool = False,
encoder: Optional[Callable[[Any], Any]] = None,
models_as_dict: bool = True,
**dumps_kwargs: Any
) -> 'unicode'
Generate a JSON representation of the model, include and exclude arguments as per dict().
encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().
key_component
class key_component(
/,
*args,
**kwargs
)
View Source
class key_component(property):
pass
Ancestors (in MRO)
- builtins.property
Class variables
fdel
fget
fset
Methods
deleter
def deleter(
...
)
Descriptor to obtain a copy of the property with a different deleter.
getter
def getter(
...
)
Descriptor to obtain a copy of the property with a different getter.
setter
def setter(
...
)
Descriptor to obtain a copy of the property with a different setter.