CRUD stands for Create, Read, Update and Delete Operations. All these Operations can be made Asynchronous using the Async Database Connection. After making Async Connection to Postgres Database, the performance of the Application improves significantly as all the operations are performed Concurrently rather than in a sequential manner. The Async database support in python is provided by Databases Library.
Databases :
Databases is a python library which gives asyncio support for various databases including PostgreSQL, MySQL, and SQLite. SQLAlchamey- the Object Relationship Mapper can be added on this Databases layer to query the database. This database support can also be integrated with any async Web-Framework for communicating with Database .
Install Databases: Run the following pip command on the terminal.
pip install databases
Install Postgresql Database Driver: Run the following pip command on the terminal.
pip install databases[postgresql]
CRUD Operations :
Initially, before we perform any operation on Database it is important to connect to database as well as set up the connection. Connecting to the database using the async function :
In the database URL, you will have to substitute the username, password, host and database for your database
Python3
from databases import Databaseimport asyncioasync def initalize_connection(): 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()) |
Output:
Connected to Database Disconnecting from Database
Create(C) : After Successful Connection to the database Let’s create a table named GfgExample using :
Python3
from databases import Databaseimport asyncioasync def create_table(): 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()) |
Output:
Connected to Database Created Table GfgExample Successfully Disconnecting from Database
Insert(I) : Now after Creation of GfgExample Table let’s insert values to it using Insert query:
Python3
from databases import Databaseimport asyncioasync def insert_records(): 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()) |
Output:
Connected to Database Inserted values in GfgExample Table Successfully Disconnecting from Database
Read(R): Now, after Insertion of values in GfgExample Table, let’s read them using Select Statement :
Python3
from databases import Databaseimport asyncioasync def find_records(): 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()) |
Output:
Connected to Database Read the values in GfgExample Table Successfully Printing Id Values Fetched from GfgExample Table 1 2 Disconnecting from Database
Delete(D): Deleting all the Records from GfgExample Table :
Python3
from databases import Databaseimport asyncioasync def delete_table(): 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()) |
Output:
Connected to Database Deleted All Records For GfgExample Successfully Disconnecting from Database
