Asynchronous Database Toolkit

0.9.0 · active · verified Sat Apr 11

The `databases` library provides asynchronous database support for Python, designed to work with `asyncio` and `await`. It supports PostgreSQL, MySQL, and SQLite, and integrates well with SQLAlchemy Core expression language. The current version is 0.9.0, and it has an active development cadence with regular updates.

Warnings

breaking Version 0.9.0 dropped support for Python 3.7 and earlier versions. Additionally, it now officially supports SQLAlchemy 2.x, which might introduce compatibility issues if you are still using older SQLAlchemy 1.x versions.
Fix: Ensure your project runs on Python 3.8 or higher. If using SQLAlchemy, consider upgrading to SQLAlchemy 2.x. If you must use older SQLAlchemy 1.x, consult `databases` documentation for specific version compatibility.
breaking In version 0.8.0, connection and transaction isolation was significantly improved. Database connections are now task-local and not inherited by child tasks. The `@db.transaction` decorator uses the calling task's connection, and new tasks use new connections unless explicitly provided.
Fix: Carefully review concurrent database operations in your application. Ensure that connections and transactions are explicitly managed or passed between tasks where shared behavior is intended, rather than relying on implicit inheritance.
gotcha You must install the specific database driver package for your chosen database alongside `databases`. For example, `pip install databases asyncpg` for PostgreSQL, `pip install databases aiomysql` for MySQL, or `pip install databases aiosqlite` for SQLite. Installing just `databases` is not sufficient for database connectivity.
Fix: Always install `databases` with the appropriate extras, e.g., `pip install databases[postgresql]`, or install the driver manually.
gotcha Compatibility with SQLAlchemy 1.4.x has been a recurring issue across several `databases` versions, with specific pins and fixes (e.g., `0.6.2` pinned `<=1.4.41`, `0.7.0` supported `>=1.4.42,<1.5`). While `0.9.0` adds SQLAlchemy 2.x support, transitioning from older SQLAlchemy 1.x with `databases` can be complex.
Fix: Always check the specific release notes for `databases` regarding SQLAlchemy compatibility. For `>=0.9.0`, it's highly recommended to use SQLAlchemy 2.x. If sticking with SQLAlchemy 1.x, careful version pinning of both `databases` and `SQLAlchemy` is required.

Install

pip install databases Base installation
pip install databases[postgresql] With PostgreSQL driver (asyncpg)
pip install databases[mysql] With MySQL driver (aiomysql)
pip install databases[sqlite] With SQLite driver (aiosqlite)

Imports

Database
```
from databases import Database
```
Record
```
from databases import Record
```
Represents a single row from a database query result.

Quickstart

This quickstart demonstrates connecting to a database, executing a DDL statement to create a table, inserting a record, and fetching all records. It uses an in-memory SQLite database by default but can be configured with an environment variable for other databases. Remember to install the appropriate database driver (e.g., `pip install databases[sqlite]`).

import asyncio
import os
from databases import Database

async def main():
    # Use an in-memory SQLite database for a simple example.
    # For a real application, use a proper URL like DATABASE_URL = 'postgresql://user:pass@host/db'
    database = Database(os.environ.get('DATABASE_URL', 'sqlite:///./test.db'))

    try:
        await database.connect()
        print("Database connected.")

        # Create a table
        query = """
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY,
                name VARCHAR(100)
            );
        """
        await database.execute(query=query)
        print("Table 'users' created or already exists.")

        # Insert data
        query = "INSERT INTO users(name) VALUES (:name)"
        await database.execute(query=query, values={"name": "Alice"})
        print("Inserted 'Alice'.")

        # Select data
        query = "SELECT id, name FROM users"
        rows = await database.fetch_all(query=query)
        for row in rows:
            print(f"User: {row['id']}, {row['name']}")

    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        await database.disconnect()
        print("Database disconnected.")

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

view raw JSON →