Database storage backends for Jobify

jobify-db provides database storage backends for the Jobify framework. It features:

• PostgreSQL support via asyncpg and psycopg drivers
• MongoDB support via motor driver
• MySQL support via aiomysql driver
• Accepts either a connection string or an externally managed pool/client
• Automatic table/collection creation on startup

Installation

Install with the driver you need:

pip install jobify-db[asyncpg]
pip install jobify-db[psycopg]
pip install jobify-db[motor]
pip install jobify-db[aiomysql]

Or with uv:

uv add "jobify-db[asyncpg]"
uv add "jobify-db[psycopg]"
uv add "jobify-db[motor]"
uv add "jobify-db[aiomysql]"

How to use

PostgreSQL with asyncpg

import asyncio

from jobify import Jobify

from jobify_db.postgresql.asyncpg import AsyncpgStorage

app = Jobify(
    storage=AsyncpgStorage(dsn="postgresql://user:password@localhost:5432/mydb"),
)


@app.task(cron="*/5 * * * *")
async def my_task() -> None:
    print("Running periodic task every 5 minutes")


async def main() -> None:
    async with app:
        await app.wait_all()


if __name__ == "__main__":
    asyncio.run(main())

PostgreSQL with psycopg

import asyncio

from jobify import Jobify

from jobify_db.postgresql.psycopg import PsycopgStorage

app = Jobify(
    storage=PsycopgStorage(
        conninfo="postgresql://user:password@localhost:5432/mydb",
    ),
)


@app.task(cron="*/1 * * * *")
async def health_check() -> None:
    print("Health check passed")


async def main() -> None:
    async with app:
        await app.wait_all()


if __name__ == "__main__":
    asyncio.run(main())

MongoDB with motor

import asyncio

from jobify import Jobify

from jobify_db.mongodb.motor import MotorStorage

app = Jobify(
    storage=MotorStorage(
        uri="mongodb://localhost:27017",
        database_name="my_app",
    ),
)


@app.task(cron="0 9 * * 1")
async def weekly_report() -> None:
    print("Generating weekly report every Monday at 9 AM")


async def main() -> None:
    async with app:
        await app.wait_all()


if __name__ == "__main__":
    asyncio.run(main())

MySQL with aiomysql

import asyncio

from jobify import Jobify

from jobify_db.mysql.aiomysql import AiomysqlStorage

app = Jobify(
    storage=AiomysqlStorage(
        host="localhost",
        port=3306,
        user="root",
        password="password",
        db="mydb",
    ),
)


@app.task(cron="*/10 * * * *")
async def sync_data() -> None:
    print("Syncing data every 10 minutes")


async def main() -> None:
    async with app:
        await app.wait_all()


if __name__ == "__main__":
    asyncio.run(main())

Using an external pool/client

You can pass an externally managed pool or client instead of a connection string. This is useful when you want to share the pool across multiple components.

import asyncio

import asyncpg
from jobify import Jobify

from jobify_db.postgresql.asyncpg import AsyncpgStorage

DSN = "postgresql://user:password@localhost:5432/mydb"


async def main() -> None:
    pool = await asyncpg.create_pool(dsn=DSN, min_size=2, max_size=20)

    app = Jobify(storage=AsyncpgStorage(pool=pool))

    @app.task(cron="0 3 * * *")
    async def nightly_cleanup() -> None:
        print("Running nightly cleanup")

    async with app:
        await app.wait_all()


if __name__ == "__main__":
    asyncio.run(main())

The same pattern works for PsycopgStorage(pool=pool), MotorStorage(client=client), and AiomysqlStorage(pool=pool).