Skip to content

anystore.model

Info

Bases: BaseModel

Streamline fs.info()

Source code in anystore/model/info.py 
47
48
49
50
51
52
53
54
55
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
class Info(BaseModel):
    """Streamline fs.info()"""

    model_config = ConfigDict(extra="ignore", populate_by_name=True)

    name: str
    """Key name: last part of the key (aka file name without path)"""

    created_at: datetime | None = Field(
        default=None,
        validation_alias=AliasChoices(*CREATED_AT_CHOICES),
    )
    """Created at timestamp"""

    updated_at: datetime | None = Field(
        default=None,
        validation_alias=AliasChoices(*UPDATED_AT_CHOICES),
    )
    """Last updated timestamp"""

    size: int
    """Size (content length) in bytes"""

    @field_validator("created_at", "updated_at", mode="before")
    @classmethod
    def coerce_timestamp(cls, v: Any) -> datetime | None:
        return _ensure_datetime(v)

    @field_validator("size", mode="before")
    @classmethod
    def ensure_size(cls, v: Any) -> int:
        if v is None:
            return 0
        return v

    @model_validator(mode="after")
    def ensure_timestamp_fallback(self) -> Self:
        """Fall back created_at <-> updated_at."""
        if self.created_at and not self.updated_at:
            self.updated_at = self.created_at
        elif self.updated_at and not self.created_at:
            self.created_at = self.updated_at
        return self

created_at = Field(default=None, validation_alias=(AliasChoices(*CREATED_AT_CHOICES))) class-attribute instance-attribute

Created at timestamp

name instance-attribute

Key name: last part of the key (aka file name without path)

size instance-attribute

Size (content length) in bytes

updated_at = Field(default=None, validation_alias=(AliasChoices(*UPDATED_AT_CHOICES))) class-attribute instance-attribute

Last updated timestamp

ensure_timestamp_fallback()

Fall back created_at <-> updated_at.

Source code in anystore/model/info.py 
82
83
84
85
86
87
88
89
@model_validator(mode="after")
def ensure_timestamp_fallback(self) -> Self:
    """Fall back created_at <-> updated_at."""
    if self.created_at and not self.updated_at:
        self.updated_at = self.created_at
    elif self.updated_at and not self.created_at:
        self.created_at = self.updated_at
    return self

Stats

Bases: Info

Meta information for a store key

Source code in anystore/model/info.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
class Stats(Info):
    """Meta information for a store key"""

    model_config = ConfigDict(extra="ignore", populate_by_name=True)

    store: str
    """Store base uri"""

    key: str
    """Full path of key"""

    mimetype: str
    """Mime type for that key"""

    @model_validator(mode="before")
    @classmethod
    def ensure_mimetype(cls, values: dict[str, Any]) -> dict[str, Any]:
        """Extract mimetype from fsspec info dict headers."""
        if not values.get("mimetype"):
            lower = {k.lower(): v for k, v in values.items()}
            mtype = lower.get("contenttype") or lower.get("mimetype")
            if mtype:
                mtype = normalize_mimetype(mtype)
                if mtype not in (DEFAULT, "binary/octet-stream"):
                    values["mimetype"] = mtype
        if not values.get("mimetype"):
            values["mimetype"] = guess_mimetype(values.get("name", ""))
        return values

    @property
    def uri(self) -> str:
        """
        Computed uri property from store uri and key.

        Returns:
            e.g. `file:///tmp/foo.txt`, `ssh://user@host:data.csv`,
                `sqlite:////tmp/db.sqlite/foo/bar`
        """
        return f"{self.store}/{self.key}"

key instance-attribute

Full path of key

mimetype instance-attribute

Mime type for that key

store instance-attribute

Store base uri

uri property

Computed uri property from store uri and key.

Returns:

Type Description
str

e.g. file:///tmp/foo.txt, ssh://user@host:data.csv, sqlite:////tmp/db.sqlite/foo/bar

ensure_mimetype(values) classmethod

Extract mimetype from fsspec info dict headers.

Source code in anystore/model/info.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@model_validator(mode="before")
@classmethod
def ensure_mimetype(cls, values: dict[str, Any]) -> dict[str, Any]:
    """Extract mimetype from fsspec info dict headers."""
    if not values.get("mimetype"):
        lower = {k.lower(): v for k, v in values.items()}
        mtype = lower.get("contenttype") or lower.get("mimetype")
        if mtype:
            mtype = normalize_mimetype(mtype)
            if mtype not in (DEFAULT, "binary/octet-stream"):
                values["mimetype"] = mtype
    if not values.get("mimetype"):
        values["mimetype"] = guess_mimetype(values.get("name", ""))
    return values

StoreModel

Bases: BaseModel

Store model to initialize a store from configuration

Source code in anystore/model/store.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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
class StoreModel(BaseModel):
    """Store model to initialize a store from configuration"""

    uri: Uri
    """Store base uri"""
    serialization_mode: Mode | None = settings.serialization_mode
    """Default serialization (auto, raw, pickle, json)"""
    serialization_func: Callable | None = None
    """Default serialization function"""
    deserialization_func: Callable | None = None
    """Default deserialization function"""
    model: Model | None = None
    """Default pydantic model for serialization"""
    raise_on_nonexist: bool | None = settings.raise_on_nonexist
    """Raise `anystore.exceptions.DoesNotExist` if key doesn't exist"""
    store_none_values: bool | None = True
    """Store `None` as value in store"""
    default_ttl: int | None = settings.default_ttl
    """Default ttl for keys (only backends that support it: redis, sql, ..)"""
    backend_config: dict[str, Any] = {}
    """Backend-specific configuration to pass through for initialization"""

    @cached_property
    def scheme(self) -> str:
        return urlparse(str(self.uri)).scheme

    @cached_property
    def path(self) -> str:
        return urlparse(str(self.uri)).path.strip("/")

    @cached_property
    def netloc(self) -> str:
        return urlparse(str(self.uri)).netloc

    @cached_property
    def is_local(self) -> bool:
        """Check if it is a local file store"""
        return self.scheme == SCHEME_FILE

    @cached_property
    def is_fslike(self) -> bool:
        """Check if it is a file-like store usable with `fsspec`"""
        return not self.is_sql and self.scheme not in (SCHEME_REDIS, SCHEME_MEMORY)

    @cached_property
    def is_http(self) -> bool:
        """Check if it is a http(s) remote store"""
        return self.scheme.startswith("http")

    @cached_property
    def is_s3(self) -> bool:
        """Check if it is a s3 (compatible) remote store"""
        return self.scheme == SCHEME_S3

    @cached_property
    def is_sql(self) -> bool:
        """Check if it is a sql-like store (sqlite, postgres, ...)"""
        return "sql" in self.scheme

    @field_validator("uri", mode="before")
    @classmethod
    def ensure_uri(cls, v: Any) -> str:
        return ensure_uri(v)

    def to_store(self, **kwargs) -> "Store":
        from anystore.store import get_store

        return get_store(**{**self.model_dump(), **kwargs})

backend_config = {} class-attribute instance-attribute

Backend-specific configuration to pass through for initialization

default_ttl = settings.default_ttl class-attribute instance-attribute

Default ttl for keys (only backends that support it: redis, sql, ..)

deserialization_func = None class-attribute instance-attribute

Default deserialization function

is_fslike cached property

Check if it is a file-like store usable with fsspec

is_http cached property

Check if it is a http(s) remote store

is_local cached property

Check if it is a local file store

is_s3 cached property

Check if it is a s3 (compatible) remote store

is_sql cached property

Check if it is a sql-like store (sqlite, postgres, ...)

model = None class-attribute instance-attribute

Default pydantic model for serialization

raise_on_nonexist = settings.raise_on_nonexist class-attribute instance-attribute

Raise anystore.exceptions.DoesNotExist if key doesn't exist

serialization_func = None class-attribute instance-attribute

Default serialization function

serialization_mode = settings.serialization_mode class-attribute instance-attribute

Default serialization (auto, raw, pickle, json)

store_none_values = True class-attribute instance-attribute

Store None as value in store

uri instance-attribute

Store base uri