Stores

In simple terms, the stores are places where, as the name suggests, stores the scheduled tasks. Asyncz has a default store wich is a MemoryStore but also brings an integration with two major ones, Redis and MongoDB.

Besides the default memory store, the tasks are not store in memory but in the corresponding desired store.

You are also capable of building your own store if you which, for instance to integrate SQLAlchemy or Tortoise ORM or anything else. Check out the custom store section for more information.

All the states are serialized and reversed using pydantic objects, so keep that in mind when creating a custom store.

MemoryStore

This is the default store when using Asyncz and probably the one you will be using when running stateless tasks or tasks that do not require some sort of cache.

Store Alias - memory

from asyncz.stores.memory import MemoryStore

RedisStore

Store Alias - redis

from asyncz.stores.redis import RedisStore

Parameters

• datababe - The database number to store tasks in.

  ^{Default: 0}

• tasks_key - The key to store tasks in.

  ^{Default: asyncz.tasks}

• run_times_key - The key to store the tasks run times in.

  ^{Default: asyncz.run_times}

• pickle_protocol - Pickle protocol level to use (for serialization), defaults to the highest available.

  ^{Default: pickle.HIGHEST_PROTOCOL}

MongoDBStore

Store Alias - mongo

from asyncz.stores.mongo import MongoDBStore

Parameters

• database - The database to store the tasks.

  ^{Default: asyncz}

• collection - The collection to store the tasks.

  ^{Default: tasks}

• client - A pymongo.mongo_client.MongoClient instance.

  ^{Default: None}

• pickle_protocol - Pickle protocol level to use (for serialization), defaults to the highest available.

  ^{Default: pickle.HIGHEST_PROTOCOL}

Custom store

from asyncz.stores.base import BaseStore

You might want to use a different store besides the ones provided and that is also possible. Foor example using a SQLAlchemy or a Tortoise ORM.

To create a custom store you must subclass the BaseStore and it should implement some other pieces of functionality as well.

from datetime import datetime
from typing import List, Union

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: Union[str, int]) -> "TaskType":
        ...

    def delete_task(self, task_id: Union[str, int]):
        ...

    def remove_all_tasks(self):
        ...

    def get_next_run_time(self) -> 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")