aioredis_fastapi

aioredis_fastapi is an asynchronous redis based session backend for FastAPI powered applications.

🚸This repository is currently under testing, kind of production-ready.🚸

🛠️ Requirements

aioredis_fastapi requires Python 3.9 or above.

To install Python 3.9, I recommend using pyenv. You can refer to this section of the readme file on how to install poetry and pyenv into your linux machine.

🚨 Installation

With pip:

python3.9 -m pip install aioredis-fastapi

or by checking out the repo and installing it with poetry:

git clone https://github.com/wiseaidev/aioredis_fastapi.git && cd aioredis_fastapi && poetry install

🚸 Usage

from typing import Any
from fastapi import Depends, FastAPI, Request, Response
from aioredis_fastapi import (
    get_session_storage,
    get_session,
    get_session_id,
    set_session,
    del_session,
    SessionStorage,
)

app = FastAPI(title=__name__)


@app.post("/set-session")
async def _set_session(
    request: Request,
    response: Response,
    session_storage: SessionStorage = Depends(get_session_storage),
):
    session_data = await request.json()
    await set_session(response, session_data, session_storage)


@app.get("/get-session")
async def _get_session(session: Any = Depends(get_session)):
    return session


@app.post("/del-session")
async def _delete_session(
    session_id: str = Depends(get_session_id),
    session_storage: SessionStorage = Depends(get_session_storage),
):
    await del_session(session_id, session_storage)
    return None

🛠️ Custom Config

from aioredis_fastapi import settings
from datetime import timedelta
import random

settings(
   redis_url="redis://localhost:6379",
   session_id_name="session-id",
   session_id_generator=lambda: str(random.randint(1000, 9999)),
   expire_time= timedelta(days=1)
)

🌐 Interacting with the endpoints

from httpx import AsyncClient
import asyncio
from aioredis_fastapi.config import settings

async def main():
    client = AsyncClient()
    r = await client.post("http://127.0.0.1:8000/set-session", json=dict(a=1, b="data", c=True))
    r = await client.get("http://127.0.0.1:8000/get-session", cookies={settings().session_id_name: "ssid"})
    print(r.text)
    return r.text

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
    loop.run_until_complete(main())
finally:
    loop.close()
    asyncio.set_event_loop(None)

🎉 Credits

The following projects were used to build and test aioredis_fastapi.

• python
• poetry
• pytest
• flake8
• coverage
• rstcheck
• mypy
• pytestcov
• tox
• isort
• black
• precommit

👋 Contribute

If you are looking for a way to contribute to the project, please refer to the Guideline.

📝 License

This program and the accompanying materials are made available under the terms and conditions of the GNU GENERAL PUBLIC LICENSE.