Operaciones CRUD en Postgres usando una base de datos asíncrona en Python

CRUD significa operaciones de creación, lectura, actualización y eliminación. Todas estas operaciones se pueden hacer asíncronas utilizando la conexión de base de datos asíncrona. Después de realizar una conexión asíncrona con la base de datos de Postgres, el rendimiento de la aplicación mejora significativamente ya que todas las operaciones se realizan simultáneamente en lugar de secuencialmente. La biblioteca de bases de datos proporciona la compatibilidad con la base de datos asíncrona en python. 

Bases de datos:

Databases es una biblioteca de python que brinda soporte asyncio para varias bases de datos, incluidas PostgreSQL, MySQL y SQLite. SQLAlchamey: el asignador de relaciones de objetos se puede agregar en esta capa de bases de datos para consultar la base de datos. Este soporte de base de datos también se puede integrar con cualquier marco web asíncrono para comunicarse con la base de datos.

Instalar bases de datos : ejecute el siguiente comando pip en la terminal.

pip install databases

Instale el controlador de la base de datos de Postgresql: ejecute el siguiente comando pip en la terminal.

pip install databases[postgresql]

Operaciones CRUD:

Inicialmente, antes de realizar cualquier operación en la base de datos, es importante conectarse a la base de datos y configurar la conexión. Conexión a la base de datos usando la función asíncrona:
en la URL de la base de datos, deberá sustituir el nombre de usuario, la contraseña, el host y la base de datos para su base de datos

from databases import Database
import asyncio
 
async def initalize_connection():
    database = Database('postgresql://username:password@host:5432/database')
    try:
        await database.connect()
        print('Connected to Database')
        await database.disconnect()
        print('Disconnecting from Database')
    except :
        print('Connection to Database Failed')
 
if __name__ == '__main__':
    asyncio.run(initalize_connection())

Producción:

Connected to Database 
Disconnecting from Database

Crear (C): después de una conexión exitosa a la base de datos, creemos una tabla llamada GfgExample usando:
 

from databases import Database
import asyncio
 
async def create_table():
 
    database = Database('postgresql://username:password@host:5432/database')
    try:
        await database.connect()
        print('Connected to Database')
 
        # Create a table.
        query = """CREATE TABLE GfgExample (id INTEGER PRIMARY KEY, name VARCHAR(100))"""
        print('Created Table GfgExample Successfully')
        await database.execute(query=query)
     
        await database.disconnect()
        print('Disconnecting from Database')
    except :
        print('Connection to Database Failed')
 
         
if __name__=='__main__':
    asyncio.run(create_table())

Producción:

Connected to Database 
Created Table GfgExample Successfully 
Disconnecting from Database

Insertar (I) : ahora, después de la creación de la tabla GfgExample , insertemos valores en ella usando Insertar consulta:
 

from databases import Database
import asyncio
 
async def insert_records():
 
    database = Database('postgresql://username:password@host:5432/database')
    try:
        await database.connect()
        print('Connected to Database')
 
        # Insert into table.
        query = """INSERT INTO GfgExample(id,name) VALUES (:id ,:name)"""
        values = [
            {"id":1,"name": "abc"},
            {"id":2,"name": "xyz"}
        ]
        await database.execute_many(query=query,values=values)
        print('Inserted values in GfgExample Table Successfully')
     
        await database.disconnect()
        print('Disconnecting from Database')
    except :
        print('Connection to Database Failed')
 
         
if __name__=='__main__':
    asyncio.run(insert_records())

Producción:

Connected to Database
Inserted values in GfgExample Table Successfully
Disconnecting from Database

Leer (R) : ahora, después de la inserción de valores en la tabla GfgExample , leámoslos usando Select Statement:

from databases import Database
import asyncio
 
async def find_records():
 
    database = Database('postgresql://username:password@host:5432/database')
    try:
        await database.connect()
        print('Connected to Database')
 
        # Select all records from table.
        query = """SELECT * FROM GfgExample"""
        rows=await database.fetch_all(query=query)
        print('Read the values in GfgExample Table Successfully')
        print('Printing Id Values Fetched from GfgExample Table')
        print(rows[0]['id'])
        print(rows[1]['id'])
     
        await database.disconnect()
        print('Disconnecting from Database')
    except :
        print('Connection to Database Failed')
 
         
if __name__=='__main__':
    asyncio.run(find_records())

Producción:

Connected to Database 
Read the values in GfgExample Table Successfully 
Printing Id Values Fetched from GfgExample Table 
1 
2 
Disconnecting from Database

Eliminar (D) : Eliminación de todos los registros de la tabla GfgExample :

from databases import Database
import asyncio
 
async def delete_table():
 
    database = Database('postgresql://username:password@host:5432/database')
    try:
        await database.connect()
        print('Connected to Database')
 
        # Delete from table.
        query = """Delete from GfgExample"""
        await database.execute(query=query)
        print('Deleted All Records For GfgExample Successfully')
 
        await database.disconnect()
        print('Disconnecting from Database')
    except :
        print('Connection to Database Failed')
         
if __name__=='__main__':
    asyncio.run(delete_table())

Producción:

Connected to Database 
Deleted All Records For GfgExample Successfully 
Disconnecting from Database