Working With Databases Using asyncio in Python — SQLAlchemy Example (79/100 Days of Python)

Day 79 of the “100 Days of Python” blog post series covering how to work with databases using asyncio

Databases are an essential part of most modern web applications, and working with them can be challenging, especially when dealing with large amounts of data. Python has several libraries for interacting with databases, including SQLAlchemy, which is one of the most popular and widely used libraries for working with databases.

Asyncio is a popular library for writing asynchronous code in Python, and it provides an efficient way to work with databases. In this tutorial, we will explore how to work with databases using asyncio and SQLAlchemy in Python.

Environment Setup

To get started, you need to set up your environment. You will need to install Python and the necessary libraries. You can use the following commands to install the required libraries:

pip install sqlalchemy[asyncio]   # Or `pip install 'sqlalchemy[asyncio]'` for zsh users
pip install aiohttp
pip install asyncpg

You can also install other libraries that you need for your project.

Working with SQLAlchemy

SQLAlchemy is a popular Python library for working with databases. It provides an ORM (Object-Relational Mapping) that allows you to work with databases using Python objects.

To use SQLAlchemy, you need to import the library and create a database connection. The following code shows how to create a connection to a PostgreSQL database using SQLAlchemy:

from sqlalchemy import create_engine
from sqlalchemy.orm import Session


engine = create_engine('postgresql://user:password@host:port/database')
# This can be `postgresql://scott:myawesomepassword@localhost:5432/mydb`
session = Session(engine, future=True)

In the above code, we create a connection to a PostgreSQL database using the create_engine function. We then create a Session object that we can use to interact with the database. Read more in the official docs: link.

Creating a Model

Once you have a connection to the database, you can create a model that represents the data you want to store in the database:

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

Base = declarative_base()


class User(Base):
    __tablename__ = 'users'

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

In the above code, we create a model for a user. We define the table name and columns using the __tablename__ and Column respectively.

Creating and Querying Data

Now that we have a model, we can create and query data using SQLAlchemy. The following code shows an example of how to create a new user and query all users:

from sqlalchemy import create_engine
from sqlalchemy.orm import Session


engine = create_engine('postgresql://user:password@host:port/database')
session = Session(engine, future=True)

# Create a new user
new_user = User(name='John Doe', email='john@gmail.com')
session.add(new_user)
session.commit()

# Query all users
users = session.execute(select(User)).all()
for user in users:
    print(user.name, user.email)

In the above code, we create a new user and add it to the database using the add function. We then commit the changes using the commit function. Finally, we query all users using the query function and print their names and emails.

Working with Asyncio and SQLAlchemy

Now that we have seen how to work with SQLAlchemy, let’s explore how to use asyncio with SQLAlchemy. To query the users we can use the following code:

import asyncio
from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

meta = MetaData()


async def main():
    engine = create_async_engine('postgresql+asyncpg://user:password@host:port/database')
    async with engine.begin() as conn:
        await conn.run_sync(meta.create_all)

    async with AsyncSession(engine) as session:
        async with session.begin():
            # Create a new user
            new_user = User(name='John Doe', email='john@gmail.com')
            session.add(new_user)

        # Query all users
        async with session.begin():
            users = await session.execute(select(User))
            async for user in users:
                print(user.name, user.email)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

In the above code, we create an async engine using the create_async_engine function from SQLAlchemy. We then create the tables using the create_all function.

To create a new user, we use the add function inside a transaction. We then query all users using the execute function inside a transaction and print their names and emails.

This tutorial provides a small example for demonstration purposes only, and there is much more to learn about databases, tables, queries, and how to work with them using libraries like SQLAlchemy and asyncio.

If you are new to databases, we recommend reading up on the basics of databases, including their structure, types, and usage. Additionally, you can explore the SQLAlchemy documentation, which provides a wealth of information on how to work with databases using Python.

Overall, this tutorial should serve as a starting point for exploring how to work with databases using asyncio and SQLAlchemy. There is much more to learn, and we encourage readers to continue exploring and experimenting with these powerful tools.

What’s next?

• If you found this story valuable, please consider clapping multiple times (this really helps a lot!)
• Hands-on Practice: Free Python Course
• Full series: 100 Days of Python
• Previous topic: Making Requests With asyncio in Python
• Next topic: Multithreading VS Multiprocessing VS Asyncio