Stores¶

Stores persist scheduled tasks. The default store is in-memory, but Asyncz also ships with durable stores for local development and production deployments.

Built-in stores¶

Store encryption¶

All built-in stores except MemoryStore support ASYNCZ_STORE_ENCRYPTION_KEY.

When it is set, Asyncz encrypts serialized task payloads before they are written to the store. This is strongly recommended for any store that can be modified by untrusted users or processes.

Persistence envelopes¶

Durable stores write a versioned task-state envelope around new task records. The envelope records the Asyncz entity kind, persistence version, selected Shape name, and native task-state payload.

Older raw TaskState pickle records still restore. Unknown Shape names or unsupported envelope versions fail clearly instead of being interpreted through another representation.

Shapes do not own store locking, encryption, task acquisition, or backend queries. Those remain store responsibilities.

MemoryStore¶

The default store. It is fast and simple, but it does not survive process restarts.

from asyncz.stores.memory import MemoryStore

FileStore¶

Stores each task as a file in a directory. This is useful for local multi-process coordination without introducing an external service.

from asyncz.stores.file import FileStore

Key parameters:

• directory
• suffix
• mode
• cleanup_directory
• pickle_protocol

RedisStore¶

from asyncz.schedulers import AsyncIOScheduler
from asyncz.stores.redis import RedisStore

scheduler = AsyncIOScheduler(stores={"default": RedisStore()})

Key parameters:

• database
• tasks_key
• run_times_key
• pickle_protocol
• host
• port

MongoDBStore¶

from asyncz.schedulers import AsyncIOScheduler
from asyncz.stores.mongo import MongoDBStore

scheduler = AsyncIOScheduler(stores={"default": MongoDBStore()})

Key parameters:

• database
• collection
• client
• pickle_protocol
• Mongo client kwargs such as host and port

SQLAlchemyStore¶

from asyncz.schedulers import AsyncIOScheduler
from asyncz.stores.sqlalchemy import SQLAlchemyStore

scheduler = AsyncIOScheduler(
    stores={"default": SQLAlchemyStore(database="sqlite:///./asyncz.sqlite3")}
)

Key parameters:

• database
• tablename
• pickle_protocol
• extra kwargs forwarded to sqlalchemy.create_engine()

Custom stores¶

If you need a different persistence backend, subclass BaseStore and implement the store contract.

from datetime import datetime
from typing import List, Union, Optional

from asyncz.schedulers import AsyncIOScheduler
from asyncz.stores.base import BaseStore
from asyncz.tasks.types import TaskType


class CustomStore(BaseStore):
    """
    A new custom store.
    """

    def get_due_tasks(self, now: datetime) -> List["TaskType"]: ...

    def lookup_task(self, task_id: str) -> "TaskType": ...

    def delete_task(self, task_id: str): ...

    def remove_all_tasks(self): ...

    def get_next_run_time(self) -> Optional[datetime]: ...

    def get_all_tasks(self) -> List["TaskType"]: ...

    def add_task(self, task: "TaskType"): ...

    def update_task(self, task: "TaskType"): ...


scheduler = AsyncIOScheduler()

# Add custom store
scheduler.add_store(CustomStore, "custom")