Let me take you inside something every Python developer has to face: making a program talk to a database. A database is just a saved file, but with superpowers. It stores rows of information, finds them quickly, and keeps them safe when the program crashes. The hard part is that your Python code and your database do not speak the same language. Your code thinks in objects and functions. The database thinks in tables and rows. The libraries in this article are the translators.

When I started with Python, I used to write manual SQL strings and paste them into every function. That worked for a week. Then I changed a table name and spent two days fixing broken code. I learned that database libraries are not optional extras. They are the bridge that keeps your code clean and your data safe. In this article, I will walk you through six Python libraries that handle most database work you will ever meet. I will show you real code, not just talk. And I will keep everything simple enough for someone who is still learning what a primary key is.

Let me start with the library that took the biggest weight off my shoulders: SQLAlchemy. SQLAlchemy is an object-relational mapper, or ORM. That means it lets you write Python classes and then turns those classes into rows in a database table. You no longer write INSERT INTO users by hand. You create a User object and add it to a session. SQLAlchemy handles the SQL for you.

Here is the smallest possible example. I use SQLite because it is built into Python and needs no server. The same code works with PostgreSQL, MySQL, and Oracle if you change the connection string.

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)

engine = create_engine('sqlite:///app.db')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

new_user = User(name='Alice')
session.add(new_user)
session.commit()

print(new_user.id)

Do you see how clean that is? I never wrote the word INSERT. I created a Python object, added it to the session, and committed. SQLAlchemy did the rest. The session pattern is the heart of the library. It keeps a copy of every object you load from the database. If you load the same row twice, you get the same Python object back. That prevents confusing bugs where one part of your program changes a user and another part still sees the old value.

You can also write queries in Python. Here is how you find every user named Alice:

from sqlalchemy import select

alice_users = session.execute(
    select(User).where(User.name == 'Alice')
).scalars().all()

for user in alice_users:
    print(user.id, user.name)

That query is safe from SQL injection because SQLAlchemy builds the SQL statement with bound parameters. You should get into the habit of using select() instead of pasting strings together. If you ever need raw SQL, SQLAlchemy lets you do that too. Its core layer builds SQL expressions that you can use without the ORM. That flexibility is why I call SQLAlchemy a toolkit, not just an ORM.

The next library is psycopg. This is the mature PostgreSQL adapter. While SQLAlchemy can talk to PostgreSQL through its own layer, psycopg is the driver under the hood. It follows a Python standard called DB-API 2.0. That standard defines how a database driver should connect, send queries, and return results. When you use psycopg, you are working close to the metal. There is no ORM mapping. You write SQL yourself, but you get parameter binding and automatic escaping.

Here is a basic psycopg connection and query:

import psycopg

with psycopg.connect("dbname=mydb user=postgres") as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT id, name FROM users WHERE name = %s", ("Alice",))
        for row in cur.fetchall():
            print(row)

Notice the strange %s placeholder. That is psycopg’s way of making sure your values are escaped safely. Never build a query like this:

# dangerous, do not do this
cur.execute(f"SELECT id, name FROM users WHERE name = '{user_input}'")

If a user enters ' OR '1'='1, they can read every row in the table. The parameterized style in the first example stops that attack. I always tell beginners: use placeholders for values, not string formatting. This one habit will save you more trouble than any other database advice.

psycopg also has features for serious data work. Server-side cursors let you run a query and then pull rows in batches. That is useful when the result has a million rows and you only want to process some of them. The library also has a connection pool, so your program can reuse connections instead of opening a new one for every request. And if you need to load a huge amount of data quickly, psycopg’s COPY support lets you stream rows straight into a table.

Now, suppose you are building a web application that handles many users at the same time. You might want asynchronous database access. That is where asyncpg comes in. asyncpg is a PostgreSQL driver that is fully async. It does not block the rest of your program while waiting for the database. It speaks the PostgreSQL wire protocol directly, without an abstraction layer, and it is fast.

Here is the same query with asyncpg:

import asyncio
import asyncpg

async def main():
    conn = await asyncpg.connect("postgresql://user:pass@localhost/mydb")
    rows = await conn.fetch("SELECT id, name FROM users WHERE name = $1", "Alice")
    for row in rows:
        print(row["id"], row["name"])
    await conn.close()

asyncio.run(main())

Do you see the $1 placeholder? asyncpg uses $1, $2, and so on. The result rows are asyncpg Record objects. You can grab values by column name, like row["name"], which is easy to read.

If you are using FastAPI or aiohttp, asyncpg fits naturally. You create a connection pool at startup, and each incoming request borrows a connection from the pool. Here is that pattern:

async def main():
    pool = await asyncpg.create_pool(
        "postgresql://user:pass@localhost/mydb",
        min_size=2,
        max_size=10
    )
    async with pool.acquire() as conn:
        result = await conn.fetchval("SELECT count(*) FROM users")
        print(result)
    await pool.close()

The pool is important because opening a database connection is slow. If every web request opens its own connection, the database will run out of sockets and memory. A pool reuses connections and makes your application handle more users with less overhead.

Let me switch to a completely different kind of database. Redis is an in-memory data store. You use it for things that need to be extremely fast: caching, temporary queues, and real-time counters. The official Python client is redis-py. It is simple to install and use. Here is the most basic example:

import redis

r = redis.Redis(host='localhost', port=6379, db=0)

r.set('greeting', 'hello')
print(r.get('greeting'))

That code stores a string in Redis and reads it back. Redis is not a relational database. It does not have tables. It has data structures like strings, hashes, lists, and sets. redis-py exposes all of them as Python methods.

Here is a slightly more useful example. I once built a rate limiter for a public API. I wanted each user to make at most ten requests per minute. Redis has a command called INCR that adds one to a number. I combined that with an expiration time.

import redis
import time

r = redis.Redis()

def is_allowed(user_id):
    key = f"rate:{user_id}:{int(time.time() // 60)}"
    count = r.incr(key)
    if count == 1:
        r.expire(key, 60)
    return count <= 10

The INCR command is atomic. That means two requests arriving at the same moment cannot both get the same number. The key includes the current minute, so the count resets every sixty seconds. This little function ran in production for months without a single bug.

redis-py also supports pipelines. A pipeline sends many commands to Redis in one round trip. That is much faster than sending them one by one. Here is an example:

pipe = r.pipeline()
pipe.set('user:1', 'Alice')
pipe.set('user:2', 'Bob')
pipe.expire('user:1', 3600)
pipe.expire('user:2', 3600)
pipe.execute()

This sends four commands with one network message. If you have a loop that performs hundreds of Redis commands, a pipeline can be the difference between slow and instant.

Now let me talk about MongoDB. MongoDB stores data as documents, not rows. The official Python driver is PyMongo. It translates Python dictionaries into BSON, a binary form of JSON. If you have worked with JSON, you already know how Mongo documents look.

Here is how you connect and insert a document:

from pymongo import MongoClient

client = MongoClient('mongodb://localhost:27017/')
db = client['shop']
collection = db['products']

product = {
    'name': 'Coffee Mug',
    'price': 12.99,
    'tags': ['glass', 'dishwasher-safe']
}
result = collection.insert_one(product)
print(result.inserted_id)

When you insert a document, MongoDB adds an _id field. PyMongo returns it so you can use it later. Reading documents is also simple:

for doc in collection.find({'tags': 'glass'}):
    print(doc['name'], doc['price'])

The find method returns a cursor. You can iterate over it like a list. If you only want one document, use find_one:

mug = collection.find_one({'name': 'Coffee Mug'})
print(mug['price'])

MongoDB also has a powerful aggregation pipeline. This is like a series of operations that filter, group, and transform documents. Here is an example that groups products by tag and counts them:

pipeline = [
    {'$unwind': '$tags'},
    {'$group': {'_id': '$tags', 'count': {'$sum': 1}}},
    {'$sort': {'count': -1}}
]
for result in collection.aggregate(pipeline):
    print(result)

PyMongo mirrors the MongoDB shell closely. If you know how to type a query in the mongo console, you can type the same dictionary in Python. That makes it easy to move between the shell and your application.

The last library on my list is Alembic. Alembic is not a database driver. It is a migration tool. A migration is a script that changes a database schema. When you add a column, rename a table, or change a column type, you need a migration. Alembic keeps track of every change and lets you apply it to different environments in order.

Alembic is made by the same people as SQLAlchemy. It reads your model metadata and compares it to the current database. Then it generates a migration file. Here is how you use it in a project.

First, you install Alembic and set it up in your project folder:

alembic init alembic

That creates an alembic directory with configuration files. Next, you connect Alembic to your database by editing alembic.ini or env.py. Then you create a migration. If your model changed, you can tell Alembic to generate a migration automatically:

alembic revision --autogenerate -m "add email to users"

Alembic looks at your SQLAlchemy model, compares it to the actual database, and writes a migration file. Here is an example of what that file contains:

from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column('users', sa.Column('email', sa.String(length=255), nullable=True))

def downgrade():
    op.drop_column('users', 'email')

The upgrade function moves the database forward. The downgrade function undoes the change. You apply the migration with this command:

alembic upgrade head

Now every environment that runs this command gets the same schema change. I have seen teams lose hours because one developer added a column to their local database and forgot to tell anyone. Alembic puts an end to that. The migration file is part of your code, so it goes through code review like any other change.

Let me show you a more realistic migration. Suppose users need a created_at timestamp. You add a column to your model, then run autogenerate. Alembic might produce this:

def upgrade():
    op.add_column(
        'users',
        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now())
    )

def downgrade():
    op.drop_column('users', 'created_at')

The server_default part tells the database to fill the column with the current time when a new row is inserted. That is often safer than setting the default in Python, because the database enforces it even when code from another language inserts a row.

Alembic also handles data changes, not just schema changes. If you want to backfill a column with data, you can run SQL inside the migration:

def upgrade():
    op.execute("UPDATE users SET email = '' WHERE email IS NULL")

But be careful. Migrations should be small and reversible. If a migration drops a table, the downgrade needs to recreate it. Otherwise you lose all data. I keep a rule: write the downgrade first in my head. If I cannot undo the change, I do not make the change.

Let me pause and talk about how these libraries work together. In many of my projects, I use SQLAlchemy as the main ORM. It talks to PostgreSQL through psycopg. When I need fast async endpoints, I use asyncpg directly. I put Redis in front of database queries to cache popular results. I store flexible product data in MongoDB. And whenever my SQLAlchemy model changes, I generate an Alembic migration to keep every developer’s database in sync.

Here is a small example of that combination. Imagine a web endpoint that shows a user’s profile. You first check Redis. If the profile is there, you return it. If not, you query PostgreSQL with asyncpg, then store the result in Redis for one minute.

import redis
import asyncpg
import asyncio

r = redis.Redis()

async def get_profile(user_id):
    cached = r.get(f"profile:{user_id}")
    if cached:
        return cached.decode('utf-8')

    conn = await asyncpg.connect(
        "postgresql://user:pass@localhost/mydb"
    )
    row = await conn.fetchrow(
        "SELECT name, email FROM users WHERE id = $1", user_id
    )
    await conn.close()

    if row:
        profile = f"{row['name']} <{row['email']}>"
        r.setex(f"profile:{user_id}", 60, profile)
        return profile
    return None

asyncio.run(get_profile(1))

This pattern saves your database from repetitive work. The first request hits PostgreSQL. The next ninety-nine requests hit Redis and finish in milliseconds. This is not a new idea, but these Python libraries make it easy to build.

You might ask which library you should learn first. I tell people to start with SQLAlchemy because it teaches you the big picture. You define models in Python, and you learn how ORMs map classes to tables. Then use psycopg or asyncpg for the times you need raw SQL. Add redis-py when you need speed. Add PyMongo when your data does not fit into strict tables. And add Alembic the moment you have more than one environment. That order has served me well.

One thing I want to stress is that you do not need to memorize every method in these libraries. I have been using SQLAlchemy for years and I still look things up. What matters is understanding the basic flow. Connect, create, read, update, delete. Each library follows that flow in its own way. Once you see the pattern, you can learn any database library quickly.

Let me show you a small CRUD example with SQLAlchemy to make the flow concrete. The C is create, the R is read, the U is update, the D is delete.

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

engine = create_engine('sqlite:///app.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# Create
u = User(name='Alice')
session.add(u)
session.commit()

# Read
user = session.query(User).filter_by(name='Alice').first()
print(user.id)

# Update
user.name = 'Alicia'
session.commit()

# Delete
session.delete(user)
session.commit()

See how the session is always in the middle. You make a change, then commit. If you forget to commit, the change disappears when the session closes. That is by design. It gives you a chance to roll back if something goes wrong. You can wrap multiple changes in one transaction:

try:
    u1 = User(name='Bob')
    u2 = User(name='Charlie')
    session.add(u1)
    session.add(u2)
    session.commit()
except Exception:
    session.rollback()

If the second insert fails, the first one is rolled back too. This is the same principle as a bank transfer: both accounts change or neither does. Learning to think in transactions is the biggest step between a beginner and a professional.

I also want to show you how to handle connection errors with psycopg. Databases go down. Networks fail. Your code should not crash in a confusing way. Here is a basic retry loop:

import psycopg
import time

def connect_with_retry():
    for attempt in range(3):
        try:
            return psycopg.connect("dbname=mydb user=postgres")
        except psycopg.OperationalError:
            print("Connection failed, retrying...")
            time.sleep(2)
    raise RuntimeError("Could not connect to database")

This is not perfect production code. For a real service, you would use a more thoughtful retry with backoff and limits. But the idea is important: assume the database will fail sometimes, and build a safe response. All six libraries let you handle errors with try and except. There is no magic.

Let me talk about performance because that is where some libraries stand out. psycopg is a blocking driver. When your program calls cur.execute, it waits for the database to respond. That is fine for many applications. asyncpg is non-blocking, so your program can do other work while waiting. redis-py is incredibly fast because Redis itself is fast. PyMongo is also asynchronous under the hood, but its default API hides that. SQLAlchemy adds some overhead because of the ORM layer, but it also gives you tools to write efficient queries. Alembic has almost no performance impact because it runs only when you apply a migration.

If you are building an API that expects thousands of requests per second, you will want asyncpg and redis-py. If you are building an internal tool that runs a report once a day, psycopg and SQLAlchemy are more than enough. Do not pick a library because someone says it is faster without testing your own workload. The best library is the one you understand and can maintain.

Let me show you a real-world use of asyncpg and redis-py in a FastAPI application. I built a simple endpoint that returns the hottest items from a list. The items live in PostgreSQL, but the top ten are cached in Redis.

from fastapi import FastAPI
import asyncpg
import redis

app = FastAPI()
cache = redis.Redis()

@app.get("/top")
async def top():
    cached = cache.get("top_items")
    if cached:
        return {"items": cached.decode()}

    conn = await asyncpg.connect(
        "postgresql://user:pass@localhost/mydb"
    )
    rows = await conn.fetch(
        "SELECT name FROM items ORDER BY score DESC LIMIT 10"
    )
    await conn.close()

    items = [row["name"] for row in rows]
    cache.setex("top_items", 30, ",".join(items))
    return {"items": items}

When the cache misses, the endpoint waits for PostgreSQL. When the cache hits, the response is almost instant. This is the standard caching pattern, and it is simple enough for a beginner to understand.

Now I want to answer a question I get a lot: should I use an ORM or write raw SQL? The answer is both. Use an ORM like SQLAlchemy for standard operations that match your Python objects. Use raw SQL for complex reports, bulk updates, and performance-critical queries. SQLAlchemy lets you switch between the two inside the same project. psycopg and asyncpg give you direct SQL when you need it. There is no moral victory in choosing one side. I use SQLAlchemy when I want to move fast, and I use raw SQL when I want exact control.

Let me also show you how PyMongo handles relationships. In a relational database, you split customers and orders into two tables and join them. In MongoDB, you often put orders inside the customer document. Here is an example:

customer = {
    'name': 'Alice',
    'orders': [
        {'item': 'Coffee Mug', 'price': 12.99},
        {'item': 'T-Shirt', 'price': 24.00}
    ]
}
db.customers.insert_one(customer)

PyMongo does not care that orders are nested. You can query for customers who bought a Coffee Mug like this:

result = db.customers.find_one({'orders.item': 'Coffee Mug'})
print(result['name'])

The dotted path orders.item tells Mongo to look inside each order in the array. This is one of the reasons people choose MongoDB for flexible data. You do not need a separate orders table with a foreign key. The whole customer record is one document.

Finally, let me go back to Alembic and show a complete workflow. This is the workflow I use on every project. First, I change my SQLAlchemy model. Then I run autogenerate. I open the generated migration file and check it. Then I upgrade.

Here is a full example. I add a last_login_at column to the User model:

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    last_login_at = Column(DateTime, nullable=True)

Then I run:

alembic revision --autogenerate -m "add last_login_at to users"

Alembic writes a file. I open it, verify the upgrade and downgrade, and run:

alembic upgrade head

Now my local database matches my code. When I push the code, the same migration runs on the staging server. Alembic tracks which migrations have run in a table called alembic_version. That table is the source of truth. If a server is behind by three migrations, running alembic upgrade head applies all three in order, as long as the migration files are present.

The thing I love about Alembic is that it turns schema changes into code review. My teammate can look at a migration and say, “Wait, you are dropping a column that still has data.” That review happens before the change reaches the database. Without migrations, someone would run a destructive command manually and only learn about the damage after it is done.

I have spent many years working with these six libraries. They are not the only database libraries in Python, but they cover the common cases. SQLAlchemy gives you a high-level way to define models. psycopg gives you a low-level way to talk to PostgreSQL. asyncpg gives you the same low-level power in an async world. redis-py makes caching and queues feel like Python dicts. PyMongo makes document storage feel like JSON. Alembic keeps your schema changes under control.

If you are new to this, start small. Install SQLite and write a few users. Then install PostgreSQL and try psycopg. Then add Redis and cache a value. You do not need to master all six at once. Learn one, use it in a small project, then add another. The code examples in this article are here for you to copy and run. Change the connection strings, change the table names, and see what happens when you break a query. That is how you learn.

One last piece of advice. Whatever library you choose, keep your database code in one place. Do not scatter queries across every file. Create a repository layer, or at least a module for each data model. That way, when you switch from psycopg to asyncpg, you only change one file. I learned that after rewriting a small app that had database calls in every route. The rewrite took one weekend. The lesson cost me that weekend. You can learn it from this article and save yourself the trouble.

Database work is not glamorous. It is detail work. But getting it right makes everything else easier. Use these six Python libraries, and you will spend less time fighting your database and more time building features that matter.