Module arti.storage.literal

None

None

View Source

from __future__ import annotations

from typing import Optional

from arti.fingerprints import Fingerprint

from arti.partitions import CompositeKey, InputFingerprints

from arti.storage import Storage, StoragePartition

_not_written_err = FileNotFoundError("Literal has not been written yet")

class StringLiteralPartition(StoragePartition):

    id: str

    value: Optional[str]

    def compute_content_fingerprint(self) -> Fingerprint:

        if self.value is None:

            raise _not_written_err

        return Fingerprint.from_string(self.value)

class StringLiteral(Storage[StringLiteralPartition]):

    """StringLiteral stores a literal String value directly in the Backend."""

    id: str = "{graph_name}/{path_tags}/{names}/{partition_key_spec}/{input_fingerprint}/{name}{extension}"

    value: Optional[str]

    def discover_partitions(

        self, input_fingerprints: InputFingerprints = InputFingerprints()

    ) -> tuple[StringLiteralPartition, ...]:

        if input_fingerprints and self.value is not None:

            raise ValueError(

                f"Literal storage cannot have a `value` preset ({self.value}) for a Producer output"

            )

        if self.key_types and not input_fingerprints:

            # We won't know what partitions to lookup.

            raise ValueError("Literal storage can only be partitioned if generated by a Producer.")

        # Existing StringLiteralPartitions may be stored in the Graph's backend, however we don't

        # have access here to lookup.

        if self.value is None:

            return ()

        return tuple(

            self.generate_partition(input_fingerprint=input_fingerprint, keys=keys)

            for keys, input_fingerprint in (

                input_fingerprints or {CompositeKey(): Fingerprint.empty()}

            ).items()

        )

Classes

StringLiteral

class StringLiteral(
    __pydantic_self__,
    **data: Any
)

View Source

class StringLiteral(Storage[StringLiteralPartition]):

    """StringLiteral stores a literal String value directly in the Backend."""

    id: str = "{graph_name}/{path_tags}/{names}/{partition_key_spec}/{input_fingerprint}/{name}{extension}"

    value: Optional[str]

    def discover_partitions(

        self, input_fingerprints: InputFingerprints = InputFingerprints()

    ) -> tuple[StringLiteralPartition, ...]:

        if input_fingerprints and self.value is not None:

            raise ValueError(

                f"Literal storage cannot have a `value` preset ({self.value}) for a Producer output"

            )

        if self.key_types and not input_fingerprints:

            # We won't know what partitions to lookup.

            raise ValueError("Literal storage can only be partitioned if generated by a Producer.")

        # Existing StringLiteralPartitions may be stored in the Graph's backend, however we don't

        # have access here to lookup.

        if self.value is None:

            return ()

        return tuple(

            self.generate_partition(input_fingerprint=input_fingerprint, keys=keys)

            for keys, input_fingerprint in (

                input_fingerprints or {CompositeKey(): Fingerprint.empty()}

            ).items()

        )

---

Ancestors (in MRO)

• arti.storage.Storage
• arti.internal.models.Model
• pydantic.main.BaseModel
• pydantic.utils.Representation
• typing.Generic

Class variables

Config

key_value_sep

partition_name_component_sep

segment_sep

storage_partition_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_orm

def from_orm(
    obj: Any
) -> 'Model'

get_default

def get_default(

) -> 'Storage[StoragePartition]'

View Source

    @classmethod

    def get_default(cls) -> Storage[StoragePartition]:

        from arti.storage.literal import StringLiteral

        return StringLiteral()  # TODO: Support some sort of configurable defaults.

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'

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

includes_input_fingerprint_template

key_types

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.

discover_partitions

def discover_partitions(
    self,
    input_fingerprints: 'InputFingerprints' = {}
) -> 'tuple[StringLiteralPartition, ...]'

View Source

    def discover_partitions(

        self, input_fingerprints: InputFingerprints = InputFingerprints()

    ) -> tuple[StringLiteralPartition, ...]:

        if input_fingerprints and self.value is not None:

            raise ValueError(

                f"Literal storage cannot have a `value` preset ({self.value}) for a Producer output"

            )

        if self.key_types and not input_fingerprints:

            # We won't know what partitions to lookup.

            raise ValueError("Literal storage can only be partitioned if generated by a Producer.")

        # Existing StringLiteralPartitions may be stored in the Graph's backend, however we don't

        # have access here to lookup.

        if self.value is None:

            return ()

        return tuple(

            self.generate_partition(input_fingerprint=input_fingerprint, keys=keys)

            for keys, input_fingerprint in (

                input_fingerprints or {CompositeKey(): Fingerprint.empty()}

            ).items()

        )

generate_partition

def generate_partition(
    self,
    keys: 'CompositeKey' = {},
    input_fingerprint: 'Fingerprint' = Fingerprint(key=None),
    with_content_fingerprint: 'bool' = True
) -> 'StoragePartitionVar_co'

View Source

    def generate_partition(

        self,

        keys: CompositeKey = CompositeKey(),

        input_fingerprint: Fingerprint = Fingerprint.empty(),

        with_content_fingerprint: bool = True,

    ) -> StoragePartitionVar_co:

        self._check_keys(self.key_types, keys)

        format_kwargs = dict[Any, Any](keys)

        if input_fingerprint.is_empty:

            if self.includes_input_fingerprint_template:

                raise ValueError(f"{self} requires an input_fingerprint, but none was provided")

        else:

            if not self.includes_input_fingerprint_template:

                raise ValueError(f"{self} does not specify a {{input_fingerprint}} template")

            format_kwargs["input_fingerprint"] = str(input_fingerprint.key)

        field_values = {

            name: (

                strip_partition_indexes(original).format(**format_kwargs)

                if lenient_issubclass(type(original := getattr(self, name)), str)

                else original

            )

            for name in self.__fields__

            if name in self.storage_partition_type.__fields__

        }

        partition = self.storage_partition_type(

            input_fingerprint=input_fingerprint, keys=keys, **field_values

        )

        if with_content_fingerprint:

            partition = partition.with_content_fingerprint()

        return partition

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().

resolve

def resolve(
    self,
    **values: 'str'
) -> 'Self'

View Source

    def resolve(self, **values: str) -> Self:

        return self.copy(

            update={

                name: new

                for name, original in self._format_fields.items()

                # Avoid "setting" the value if not updated to reduce pydantic repr verbosity (which

                # only shows "set" fields by default).

                if (new := self._resolve_field(name, original, values)) != original

            }

        )

StringLiteralPartition

class StringLiteralPartition(
    __pydantic_self__,
    **data: Any
)

View Source

class StringLiteralPartition(StoragePartition):

    id: str

    value: Optional[str]

    def compute_content_fingerprint(self) -> Fingerprint:

        if self.value is None:

            raise _not_written_err

        return Fingerprint.from_string(self.value)

---

Ancestors (in MRO)

• arti.storage.StoragePartition
• arti.internal.models.Model
• pydantic.main.BaseModel
• pydantic.utils.Representation

Class variables

Config

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_orm

def from_orm(
    obj: Any
) -> 'Model'

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'

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

compute_content_fingerprint

def compute_content_fingerprint(
    self
) -> 'Fingerprint'

View Source

    def compute_content_fingerprint(self) -> Fingerprint:

        if self.value is None:

            raise _not_written_err

        return Fingerprint.from_string(self.value)

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().

with_content_fingerprint

def with_content_fingerprint(
    self,
    keep_existing: 'bool' = True
) -> 'Self'

View Source

    def with_content_fingerprint(self, keep_existing: bool = True) -> Self:

        if keep_existing and not self.content_fingerprint.is_empty:

            return self

        return self.copy(update={"content_fingerprint": self.compute_content_fingerprint()})