Skip to content

anystore.model

Models

Pydantic model interfaces to initialize stores and handle metadata for keys.

BaseStats

Bases: BaseModel

Shared base metadata object

Source code in anystore/model.py 
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
class BaseStats(BaseModel):
    """Shared base metadata object"""

    created_at: datetime | None = None
    """Created at timestamp"""

    updated_at: datetime | None = None
    """Last updated timestamp"""

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

    raw: SDict = {}
    """Raw data (to preserve headers)"""

    @field_validator("size", mode="before")
    @classmethod
    def ensure_size(cls, value: Any) -> Any:
        return value or 0

    @model_validator(mode="after")
    def ensure_updated_at(self) -> Self:
        self.updated_at = self.updated_at or self.created_at
        return self

    @model_validator(mode="after")
    def ensure_created_at(self) -> Self:
        self.created_at = self.created_at or self.updated_at
        return self

created_at = None class-attribute instance-attribute

Created at timestamp

raw = {} class-attribute instance-attribute

Raw data (to preserve headers)

size instance-attribute

Size (content length) in bytes

updated_at = None class-attribute instance-attribute

Last updated timestamp

Stats

Bases: BaseStats

Meta information for a store key

Source code in anystore/model.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class Stats(BaseStats):
    """Meta information for a store key"""

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

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

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

    @property
    def uri(self) -> str:
        """
        Computed uri property. Absolute when file-like prepended with store
        schema, relative if using different store backend

        Returns:
            file-like: `file:///tmp/foo.txt`, `ssh://user@host:data.csv`
            relative path for other (redis, sql, ...): `tmp/foo.txt`
        """
        store = StoreModel(uri=self.store)
        if store.is_fslike:
            return join_uri(self.store, self.key)
        return self.key

    @property
    def mimetype(self) -> str:
        """
        Return the mimetype based on response headers or extension
        """
        if self.raw:
            mtype = self.raw.get("ContentType") or self.raw.get("mimetype")
            mtype = normalize_mimetype(mtype)
            if mtype not in (DEFAULT, "binary/octet-stream"):
                return mtype
        return guess_mimetype(self.name)

key instance-attribute

Full path of key

mimetype property

Return the mimetype based on response headers or extension

name instance-attribute

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

store instance-attribute

Store base uri

uri property

Computed uri property. Absolute when file-like prepended with store schema, relative if using different store backend

Returns:

Type Description
str

file-like: file:///tmp/foo.txt, ssh://user@host:data.csv
str

relative path for other (redis, sql, ...): tmp/foo.txt

StoreModel

Bases: BaseModel

Store model to initialize a store from configuration

Source code in anystore/model.py 
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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
class StoreModel(BaseModel):
    """Store model to initialize a store from configuration"""

    uri: Uri
    """Store base uri"""
    key_prefix: str | None = None
    """Global key prefix for all keys"""
    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"""
    readonly: bool | None = False
    """Consider this store as a read-only store, writing will raise an exception"""

    @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:
        uri = ensure_uri(v)
        return uri.rstrip("/")

    def to_store(self) -> "BaseStore":
        from anystore.store import get_store

        return get_store(**self.model_dump())

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

key_prefix = None class-attribute instance-attribute

Global key prefix for all keys

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

readonly = False class-attribute instance-attribute

Consider this store as a read-only store, writing will raise an exception

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