在Python中使用异步数据库对 Postgres 进行 CRUD 操作

CRUD 代表创建、读取、更新和删除操作。所有这些操作都可以使用异步数据库连接进行异步操作。与 Postgres 数据库建立异步连接后，应用程序的性能显着提高，因为所有操作都是同时执行的，而不是以顺序方式执行的。 Python中的异步数据库支持由 Databases Library 提供。

数据库：

Databases 是一个Python库，它为各种数据库提供异步支持，包括 PostgreSQL、MySQL 和 SQLite。 SQLAlchamey - 可以在此数据库层上添加对象关系映射器以查询数据库。这种数据库支持也可以与任何异步 Web 框架集成，以便与 Database 进行通信。

安装数据库：在终端上运行以下 pip 命令。

pip install databases

安装 Postgresql 数据库驱动程序：在终端上运行以下 pip 命令。

pip install databases[postgresql]

CRUD 操作：

最初，在我们对数据库执行任何操作之前，连接到数据库以及设置连接非常重要。使用异步函数连接到数据库：
在数据库 URL 中，您必须将用户名、密码、主机和数据库替换为您的数据库

Python3

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())

输出：

Connected to Database 
Disconnecting from Database

Create(C) :成功连接到数据库后让我们使用以下命令创建一个名为GfgExample的表：

Python3

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())

输出：

Connected to Database 
Created Table GfgExample Successfully 
Disconnecting from Database

Insert(I) ：现在在创建GfgExample表之后，让我们使用 Insert 查询向其插入值：

Python3

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())

输出：

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

Read(R) ：现在，在GfgExample表中插入值之后，让我们使用 Select 语句读取它们：

Python3

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())

输出：

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

Delete(D) ：从GfgExample表中删除所有记录：

Python3

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())

输出：

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