harnest.assets

All Python modules · Import and compatibility guidance

Exports

Name	Kind
`AssetMediaMetadata`	Class
`AssetNotFoundError`	Class
`AssetQuotaError`	Class
`AssetRecord`	Class
`AssetScope`	Class
`AssetStorage`	Class
`AssetStore`	Class
`AssetStoreError`	Class
`AssetURLStorage`	Class
`AssetValidationError`	Class
`DEFAULT_MAX_ASSET_BYTES`	Value or type alias
`DEFAULT_MAX_TOTAL_ASSET_BYTES`	Value or type alias
`MemoryAssetStore`	Class
`Stored`	Class

AssetMediaMetadata

Store inspected, non-content media properties safe for policy checks.

AssetMediaMetadata(
    width: int | None = None,
    height: int | None = None,
    duration_seconds: float | None = None,
    page_count: int | None = None,
    frame_count: int | None = None,
    channel_count: int | None = None,
    sample_rate_hz: int | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`width`	`int \| None`	`None`
`height`	`int \| None`	`None`
`duration_seconds`	`float \| None`	`None`
`page_count`	`int \| None`	`None`
`frame_count`	`int \| None`	`None`
`channel_count`	`int \| None`	`None`
`sample_rate_hz`	`int \| None`	`None`

Parameters and example

Parameter	Type	Default
`width`	`int \| None`	`None`
`height`	`int \| None`	`None`
`duration_seconds`	`float \| None`	`None`
`page_count`	`int \| None`	`None`
`frame_count`	`int \| None`	`None`
`channel_count`	`int \| None`	`None`
`sample_rate_hz`	`int \| None`	`None`

from harnest.assets import AssetMediaMetadata

metadata = AssetMediaMetadata()

AssetNotFoundError

Indicate that an asset is absent or inaccessible to the caller.

Example

from harnest.assets import AssetNotFoundError

try:
    ...
except AssetNotFoundError as error:
    ...

AssetQuotaError

Indicate that an asset or store capacity limit was exceeded.

Example

from harnest.assets import AssetQuotaError

try:
    ...
except AssetQuotaError as error:
    ...

AssetRecord

Describe a stored asset without exposing its binary content.

AssetRecord(
    asset_id: str,
    scope: AssetScope,
    media_type: str,
    size_bytes: int,
    created_at: datetime,
    expires_at: datetime | None = None,
    metadata: AssetMediaMetadata | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`asset_id`	`str`	`—`
`scope`	`AssetScope`	`—`
`media_type`	`str`	`—`
`size_bytes`	`int`	`—`
`created_at`	`datetime`	`—`
`expires_at`	`datetime \| None`	`None`
`metadata`	`AssetMediaMetadata \| None`	`None`

Parameters and example

Parameter	Type	Default
`asset_id`	`str`	Required
`scope`	`AssetScope`	Required
`media_type`	`str`	Required
`size_bytes`	`int`	Required
`created_at`	`datetime`	Required
`expires_at`	`datetime \| None`	`None`
`metadata`	`AssetMediaMetadata \| None`	`None`

from harnest.assets import AssetRecord

record = AssetRecord(
    asset_id=...,
    scope=...,
    media_type=...,
    size_bytes=...,
    created_at=...,
)

AssetScope

Bind an asset to one authenticated user and session.

AssetScope(user_id: str, session_id: str) -> None

Fields (— means no declared default):

Field	Type	Default
`user_id`	`str`	`—`
`session_id`	`str`	`—`

Parameters and example

Parameter	Type	Default
`user_id`	`str`	Required
`session_id`	`str`	Required

from harnest.assets import AssetScope

scope = AssetScope(user_id=..., session_id=...)

AssetStorage

Persist and retrieve binary assets behind scoped opaque references.

AssetStorage(*args, **kwargs)

Example

from harnest.assets import AssetStorage

class MyAssetStorage(AssetStorage):
    ...

AssetStorage.start

async AssetStorage.start() -> None

Initialize resources required by the asset store.

Example

from harnest.assets import AssetStorage

storage: AssetStorage
await storage.start()

AssetStorage.save

async AssetStorage.save(
    *,
    scope: AssetScope,
    media_type: str,
    chunks: AsyncIterable[bytes],
    metadata: AssetMediaMetadata | None = None,
    retention_seconds: float | None = None,
    path: str | None = None,
) -> AssetRecord

Persist streamed bytes and return their scoped asset record.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required
`media_type`	`str`	Required
`chunks`	`AsyncIterable[bytes]`	Required
`metadata`	`AssetMediaMetadata \| None`	`None`
`retention_seconds`	`float \| None`	`None`
`path`	`str \| None`	`None`

from harnest.assets import AssetStorage

storage: AssetStorage
record = await storage.save(scope=..., media_type=..., chunks=...)

AssetStorage.stat

async AssetStorage.stat(*, scope: AssetScope, asset_id: str) -> AssetRecord | None

Return metadata for an owned asset, or None when unavailable.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required
`asset_id`	`str`	Required

from harnest.assets import AssetStorage

storage: AssetStorage
record = await storage.stat(scope=..., asset_id=...)

AssetStorage.open

AssetStorage.open(*, scope: AssetScope, asset_id: str) -> AsyncIterator[bytes]

Stream the bytes for an asset owned by the supplied scope.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required
`asset_id`	`str`	Required

from harnest.assets import AssetStorage

storage: AssetStorage
result = storage.open(scope=..., asset_id=...)

AssetStorage.delete

async AssetStorage.delete(*, scope: AssetScope, asset_id: str) -> bool

Delete an owned asset and report whether one was removed.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required
`asset_id`	`str`	Required

from harnest.assets import AssetStorage

storage: AssetStorage
result = await storage.delete(scope=..., asset_id=...)

AssetStorage.delete_scope

async AssetStorage.delete_scope(*, scope: AssetScope) -> int

Delete every asset in a scope and return the removal count.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required

from harnest.assets import AssetStorage

storage: AssetStorage
result = await storage.delete_scope(scope=...)

AssetStorage.close

async AssetStorage.close() -> None

Release resources owned by the asset store.

Example

from harnest.assets import AssetStorage

storage: AssetStorage
await storage.close()

AssetStore

Persist and retrieve binary assets behind scoped opaque references.

AssetStore(*args, **kwargs)

Example

from harnest.assets import AssetStore

class MyAssetStore(AssetStore):
    ...

AssetStore.start

async AssetStore.start() -> None

Initialize resources required by the asset store.

Example

from harnest.assets import AssetStore

store: AssetStore
await store.start()

AssetStore.save

async AssetStore.save(
    *,
    scope: AssetScope,
    media_type: str,
    chunks: AsyncIterable[bytes],
    metadata: AssetMediaMetadata | None = None,
    retention_seconds: float | None = None,
    path: str | None = None,
) -> AssetRecord

Persist streamed bytes and return their scoped asset record.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required
`media_type`	`str`	Required
`chunks`	`AsyncIterable[bytes]`	Required
`metadata`	`AssetMediaMetadata \| None`	`None`
`retention_seconds`	`float \| None`	`None`
`path`	`str \| None`	`None`

from harnest.assets import AssetStore

store: AssetStore
record = await store.save(scope=..., media_type=..., chunks=...)

AssetStore.stat

async AssetStore.stat(*, scope: AssetScope, asset_id: str) -> AssetRecord | None

Return metadata for an owned asset, or None when unavailable.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required
`asset_id`	`str`	Required

from harnest.assets import AssetStore

store: AssetStore
record = await store.stat(scope=..., asset_id=...)

AssetStore.open

AssetStore.open(*, scope: AssetScope, asset_id: str) -> AsyncIterator[bytes]

Stream the bytes for an asset owned by the supplied scope.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required
`asset_id`	`str`	Required

from harnest.assets import AssetStore

store: AssetStore
result = store.open(scope=..., asset_id=...)

AssetStore.delete

async AssetStore.delete(*, scope: AssetScope, asset_id: str) -> bool

Delete an owned asset and report whether one was removed.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required
`asset_id`	`str`	Required

from harnest.assets import AssetStore

store: AssetStore
result = await store.delete(scope=..., asset_id=...)

AssetStore.delete_scope

async AssetStore.delete_scope(*, scope: AssetScope) -> int

Delete every asset in a scope and return the removal count.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required

from harnest.assets import AssetStore

store: AssetStore
result = await store.delete_scope(scope=...)

AssetStore.close

async AssetStore.close() -> None

Release resources owned by the asset store.

Example

from harnest.assets import AssetStore

store: AssetStore
await store.close()

AssetStoreError

Base error for asset-store operations without sensitive context.

Example

from harnest.assets import AssetStoreError

try:
    ...
except AssetStoreError as error:
    ...

AssetURLStorage

Optional capability for model-call-time temporary URL generation.

AssetURLStorage(*args, **kwargs)

Example

from harnest.assets import AssetURLStorage

class MyAssetURLStorage(AssetURLStorage):
    ...

AssetURLStorage.signed_url

async AssetURLStorage.signed_url(*, scope: AssetScope, asset_id: str, expires_in: float) -> str

Create a short-lived URL for an asset owned by the supplied scope.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required
`asset_id`	`str`	Required
`expires_in`	`float`	Required

from harnest.assets import AssetURLStorage

storage: AssetURLStorage
result = await storage.signed_url(scope=..., asset_id=..., expires_in=...)

AssetValidationError

Indicate malformed asset input without reflecting the rejected value.

Example

from harnest.assets import AssetValidationError

try:
    ...
except AssetValidationError as error:
    ...

DEFAULT_MAX_ASSET_BYTES

DEFAULT_MAX_ASSET_BYTES = 10485760

Example

from harnest.assets import DEFAULT_MAX_ASSET_BYTES

DEFAULT_MAX_TOTAL_ASSET_BYTES

DEFAULT_MAX_TOTAL_ASSET_BYTES = 104857600

Example

from harnest.assets import DEFAULT_MAX_TOTAL_ASSET_BYTES

MemoryAssetStore

Bounded, process-local asset storage intended for development and tests.

MemoryAssetStore(
    *,
    max_asset_bytes: int = 10485760,
    max_total_bytes: int = 104857600,
    read_chunk_bytes: int = 65536,
    default_retention_seconds: float | None = None,
    clock: Callable[[], datetime] | None = None,
    token_factory: Callable[[], str] | None = None,
) -> None

Parameters and example

Parameter	Type	Default
`max_asset_bytes`	`int`	`10485760`
`max_total_bytes`	`int`	`104857600`
`read_chunk_bytes`	`int`	`65536`
`default_retention_seconds`	`float \| None`	`None`
`clock`	`Callable[[], datetime] \| None`	`None`
`token_factory`	`Callable[[], str] \| None`	`None`

from harnest.assets import MemoryAssetStore

store = MemoryAssetStore()

MemoryAssetStore.total_bytes

Report committed bytes for development diagnostics.

Example

from harnest.assets import MemoryAssetStore

store: MemoryAssetStore
value = store.total_bytes

MemoryAssetStore.start

async MemoryAssetStore.start() -> None

Satisfy lifecycle ownership without allocating external resources.

Example

from harnest.assets import MemoryAssetStore

store: MemoryAssetStore
await store.start()

MemoryAssetStore.save

async MemoryAssetStore.save(
    *,
    scope: AssetScope,
    media_type: str,
    chunks: AsyncIterable[bytes],
    metadata: AssetMediaMetadata | None = None,
    retention_seconds: float | None = None,
    path: str | None = None,
) -> AssetRecord

Stream an asset into reserved capacity and publish it atomically.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required
`media_type`	`str`	Required
`chunks`	`AsyncIterable[bytes]`	Required
`metadata`	`AssetMediaMetadata \| None`	`None`
`retention_seconds`	`float \| None`	`None`
`path`	`str \| None`	`None`

from harnest.assets import MemoryAssetStore

store: MemoryAssetStore
record = await store.save(scope=..., media_type=..., chunks=...)

MemoryAssetStore.stat

async MemoryAssetStore.stat(*, scope: AssetScope, asset_id: str) -> AssetRecord | None

Return metadata only when the caller owns an unexpired asset.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required
`asset_id`	`str`	Required

from harnest.assets import MemoryAssetStore

store: MemoryAssetStore
record = await store.stat(scope=..., asset_id=...)

MemoryAssetStore.open

MemoryAssetStore.open(*, scope: AssetScope, asset_id: str) -> AsyncIterator[bytes]

Yield a stable snapshot while later deletion blocks new readers.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required
`asset_id`	`str`	Required

from harnest.assets import MemoryAssetStore

store: MemoryAssetStore
async for item in store.open(scope=..., asset_id=...):
    ...

MemoryAssetStore.delete

async MemoryAssetStore.delete(*, scope: AssetScope, asset_id: str) -> bool

Delete an owned asset without revealing cross-principal existence.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required
`asset_id`	`str`	Required

from harnest.assets import MemoryAssetStore

store: MemoryAssetStore
result = await store.delete(scope=..., asset_id=...)

MemoryAssetStore.delete_scope

async MemoryAssetStore.delete_scope(*, scope: AssetScope) -> int

Remove all assets owned by a deleted session without cross-scope reads.

Parameters and example

Parameter	Type	Default
`scope`	`AssetScope`	Required

from harnest.assets import MemoryAssetStore

store: MemoryAssetStore
result = await store.delete_scope(scope=...)

MemoryAssetStore.cleanup_expired

async MemoryAssetStore.cleanup_expired() -> int

Synchronously remove expired assets and return the removal count.

Example

from harnest.assets import MemoryAssetStore

store: MemoryAssetStore
result = await store.cleanup_expired()

MemoryAssetStore.clear

async MemoryAssetStore.clear() -> None

Deterministically discard all development-store content.

Example

from harnest.assets import MemoryAssetStore

store: MemoryAssetStore
await store.clear()

MemoryAssetStore.close

async MemoryAssetStore.close() -> None

Discard process-local bytes when the owning application stops.

Example

from harnest.assets import MemoryAssetStore

store: MemoryAssetStore
await store.close()

Stored

Persist a media field using one explicitly named asset storage. expires_in controls the model-facing URL lifetime. retention controls stored-byte lifetime; the two durations are deliberately independent.

Stored(
    store: str = 'default',
    path: str | None = None,
    expires_in: float = 60,
    retention: timedelta | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`store`	`str`	`'default'`
`path`	`str \| None`	`None`
`expires_in`	`float`	`60`
`retention`	`timedelta \| None`	`None`

Parameters and example

Parameter	Type	Default
`store`	`str`	`'default'`
`path`	`str \| None`	`None`
`expires_in`	`float`	`60`
`retention`	`timedelta \| None`	`None`

from harnest.assets import Stored

stored = Stored()

Stored.retention_seconds

Return the store contract’s numeric retention representation.

Example

from harnest.assets import Stored

stored: Stored
value = stored.retention_seconds

​Exports

​AssetMediaMetadata

​AssetNotFoundError

​AssetQuotaError

​AssetRecord

​AssetScope

​AssetStorage

​AssetStorage.start

​AssetStorage.save

​AssetStorage.stat

​AssetStorage.open

​AssetStorage.delete

​AssetStorage.delete_scope

​AssetStorage.close

​AssetStore

​AssetStore.start

​AssetStore.save

​AssetStore.stat

​AssetStore.open

​AssetStore.delete

​AssetStore.delete_scope

​AssetStore.close

​AssetStoreError

​AssetURLStorage

​AssetURLStorage.signed_url

​AssetValidationError

​DEFAULT_MAX_ASSET_BYTES

​DEFAULT_MAX_TOTAL_ASSET_BYTES

​MemoryAssetStore

​MemoryAssetStore.total_bytes

​MemoryAssetStore.start

​MemoryAssetStore.save

​MemoryAssetStore.stat

​MemoryAssetStore.open

​MemoryAssetStore.delete

​MemoryAssetStore.delete_scope

​MemoryAssetStore.cleanup_expired

​MemoryAssetStore.clear

​MemoryAssetStore.close

​Stored

​Stored.retention_seconds

Exports

AssetMediaMetadata

AssetNotFoundError

AssetQuotaError

AssetRecord

AssetScope

AssetStorage

AssetStorage.start

AssetStorage.save

AssetStorage.stat

AssetStorage.open

AssetStorage.delete

AssetStorage.delete_scope

AssetStorage.close

AssetStore

AssetStore.start

AssetStore.save

AssetStore.stat

AssetStore.open

AssetStore.delete

AssetStore.delete_scope

AssetStore.close

AssetStoreError

AssetURLStorage

AssetURLStorage.signed_url

AssetValidationError

DEFAULT_MAX_ASSET_BYTES

DEFAULT_MAX_TOTAL_ASSET_BYTES

MemoryAssetStore

MemoryAssetStore.total_bytes

MemoryAssetStore.start

MemoryAssetStore.save

MemoryAssetStore.stat

MemoryAssetStore.open

MemoryAssetStore.delete

MemoryAssetStore.delete_scope

MemoryAssetStore.cleanup_expired

MemoryAssetStore.clear

MemoryAssetStore.close

Stored

Stored.retention_seconds