Home
Essentials
Features
And more
Community
Project
Documentation Blog Discussions Issues
Back to the blog
Releases

Announcing Expanse 0.7.0

TE
The Expanse Team

The Expanse team is pleased to announce the immediate availability of Expanse 0.7.0, which includes several new features and improvements like a new jobs system, a message bus, and caching.

New jobs system

Expanse now includes a new jobs system that allows you to run background tasks in your application. The new jobs system supports both synchronous and asynchronous jobs, and provides a simple and intuitive API for defining and executing jobs.

You can create a new job using the make job command of the CLI:

expanse make job ProcessFileUpload

expanse make job ProcessFileUpload --sync

This will generate a new job class in the app.jobs.process_file_upload module. You can then define the job payload and implement the execute method to perform the desired task.

from dataclasses import dataclass

from expanse.jobs.asynchronous.job import Job


@dataclass
class ProcessFileUploadPayload:

    file_path: str


class ProcessFileUpload(Job[ProcessFileUploadPayload]):

    async def execute(self) -> None:
        file_path = payload.file_path

        # Process the file
        ...

from dataclasses import dataclass

from expanse.jobs.synchronous.job import Job


@dataclass
class ProcessFileUploadPayload:

    file_path: str


class ProcessFileUpload(Job[ProcessFileUploadPayload]):

    def execute(self) -> None:
        file_path = self.payload.file_path

        # Process the file
        ...

Once you have created your job, you can dispatch it using the dispatch() (or dispatch_sync()) method.

from expanse.http.upload_file import UploadFile
from expanse.jobs.job_dispatcher import JobDispatcher

from app.jobs.process_file_upload import ProcessFileUpload, ProcessFileUploadPayload


async def upload_file(
    file_: UploadFile
) -> Response:
    file_path = await file_.save("files")

    await ProcessFileUpload(
        ProcessFileUploadPayload(file_path=file_path)
    ).dispatch()

    ...

from expanse.http.upload_file import UploadFile
from expanse.jobs.job_dispatcher import JobDispatcher

from app.jobs.process_file_upload import ProcessFileUpload, ProcessFileUploadPayload


def upload_file(
    file_: UploadFile
) -> Response:
    file_path = file_.save_sync("files")

    ProcessFileUpload(
        ProcessFileUploadPayload(file_path=file_path)
    ).dispatch_sync()

    ...

For more information about the new jobs system, please refer to the documentation.

Message bus

Expanse now includes a message bus that allows you to send and receive messages asynchronously between different parts of your application. This is useful for decoupling different parts of your application and for implementing event-driven architectures. The message bus powers the new jobs system, but can also be used independently.

Creating messages and handlers

A message is a simple class that can be serialized to JSON. The message bus natively supports dataclasses, Pydantic models and msgspec structs.

from dataclasses import dataclass


@dataclass
class UserCreated:

    user_id: int
    username: str

from pydantic import BaseModel


class UserCreated(BaseModel):

    user_id: int
    username: str

import msgspec


class UserCreated(msgspec.Struct):

    user_id: int
    username: str

A message handler is any callable that takes a message as an argument, marked with the @message_handler decorator.

from expanse.messenger.utils import message_handler

from app.messages.user_created import UserCreated


@message_handler()
async def send_welcome_email(message: UserCreated) -> None:
    ...

By default, Expanse will look for message handlers in the app/message_handlers directory.

Dispatching messages

To dispatch a message, you may inject an instance of the MessageBus and call the dispatch method.

from expanse.messenger.asynchronous.message_bus import MessageBus

from app.messages.user_created import UserCreated


class UserController:

    async def create_user(self, bus: MessageBus) -> Response:
        ...
        await bus.dispatch(UserCreated(user_id=1, username="john_doe"))

        return Response(status_code=201)
        ...

from expanse.messenger.synchronous.message_bus import MessageBus

from app.messages.user_created import UserCreated


class UserController:

    def create_user(self, bus: MessageBus) -> Response:
        ...
        bus.dispatch(UserCreated(user_id=1, username="john_doe"))

        return Response(status_code=201)
        ...

Caching

Expanse now provides a cache system with powerful features at your disposal:

Cache stores

The cache system is configured based on two concepts: drivers and stores. Drivers are the underlying implementations of the cache system, like Redis or a database, while stores represent a caching layer supported by one of the supported drivers. You can have multiple stores using the same driver, and each store can be configured independently.

To start using the cache system, you need to configure at least one store and its corresponding driver and reference it in the CACHE_STORE environment variable.

CACHE_STORE=default

Each store is configured via environment variables following the pattern CACHE_STORES__STORE_NAME__PARAMETER=VALUE, where STORE_NAME is the name of the store and PARAMETER is a parameter supported by the driver used by the store.

Out of the box, Expanse supports the following drivers:

  • redis: Uses Redis as the underlying cache store. To use the Redis driver, you need to have at least one Redis connection configured.
  • database: Uses a database table as the underlying cache store. To use the Database driver, you need to have at least one database connection configured.
  • file: Uses the file system as the underlying cache store.
  • memory: Uses an LRU (Least Recently Used) in-memory cache as the underlying cache store. This driver is best suited for L1 caches in a multi-tier caching system.

You can refer to the documentation for more information about configuring cache stores.

Using the cache

To start using the cache in your application, you can inject a Cache instance into your routes, controllers, or any other part of your application.

from expanse.contracts.cache.asynchronous.cache import Cache


async def my_route(cache: Cache) -> Response:
    await cache.set("key", "value", 3600)

    return Response()

from expanse.contracts.cache.synchronous.cache import Cache


def my_route(cache: Cache) -> Response:
    cache.set("key", "value", 3600)

    return Response()

The preferred way to retrieve and store items in the cache is by using the remember method. It tries to retrieve the item from the cache, and if it does not exist or has expired, it computes the value using a callback function, stores it in the cache, and returns it.

from typing import Sequence
from functools import partial
from sqlalchemy import select

from expanse.contracts.cache.asynchronous.cache import Cache
from expanse.database.asynchronous.session import AsyncSession
from expanse.http.response import Response
from expanse.http.helpers import json

from app.models.user import User


async def get_users(session: AsyncSession) -> Sequence[User]:
    return (await session.execute(select(User))).scalars().all()


async def retrieve_users(cache: Cache, session: AsyncSession) -> Response:
    users = await cache.remember("key", partial(get_users, session), 600)

    return json([{"id": user.id, "name": user.name} for user in users])

from typing import Sequence
from functools import partial
from sqlalchemy import select

from expanse.contracts.cache.synchronous.cache import Cache
from expanse.database.synchronous.session import Session
from expanse.http.response import Response
from expanse.http.helpers import json

from app.models.user import User


def get_users(session: Session) -> Sequence[User]:
    return session.execute(select(User)).scalars().all()


def retrieve_users(cache: Cache, session: Session) -> Response:
    users = cache.remember("key", partial(get_users, session), 600)

    return json([{"id": user.id, "name": user.name} for user in users])

The remember method accepts three parameters: the cache key, a callback function to compute the value if it does not exist or has expired, and the time-to-live (TTL) of the cache item in seconds. The TTL can be omitted if you want the item to never expire.

The main advantage of using the remember method is that it provides stampede protection out of the box.

Multi-tier caching

Multi-tier caching is a caching strategy that uses multiple cache stores to improve the performance of your application. The most common use case for multi-tier caching is to use an in-memory cache as a first layer (L1 cache) and a persistent cache as a second layer (L2 cache).

When using multi-tier caching, the cache system will first try to retrieve the item from the fast, in-memory L1 cache. If it does not exist or has expired, it will try to retrieve it from the L2 (Redis for instance) cache. If the item exists in the L2 cache and is still valid, it will be stored in the L1 cache and returned.

When items are stored or deleted, both cache layers are updated and a bus notifies other running application instances to evict the items from their L1 cache.

To learn more about multi-tier caching, please refer to the documentation.

Stampede protection

Cache stampede (also known as cache thundering) is a situation that occurs when multiple requests try to access a cache item that does not exist or has expired at the same time. This can lead to a sudden spike in traffic to the underlying data source, which can cause performance issues or even downtime.

Thankfully, Expanse provides cache stampede protection out of the box when using the remember method to retrieve items from the cache. When the first request tries to retrieve an item that does not exist or has expired, it will acquire a lock for that cache key and compute the value using the provided callback function. Meanwhile, any other request that tries to retrieve the same item will wait for the lock to be released and then try to retrieve the item from the cache again. This way, only one request will compute the value for a given cache key at a time, regardless of how many requests are processed at the same time.