# 您可以通过将数据库用作异步上下文管理器来控制数据库的连接/断开连接。
import config
print(config.DATABASE_URL)
from databases import Database
async with Database(config.DATABASE_URL) as database:
row = await database.fetch_one(query="select * from test_user;")
print(row)
'''
If you're integrating against a web framework,
then you'll probably want to hook into framework
startup or shutdown events. For example, with Starlette you would use the following:
'''
@app.on_event("startup")
async def startup():
await database.connect()
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()
# 连接选项
# Use an SSL connection.
database = Database('postgresql://localhost/example?ssl=true')
# Use a connection pool of between 5-20 connections.
database = Database('mysql://localhost/example?min_size=5&max_size=20')
# 使用键值对设置选项
database = Database('postgresql://localhost/example', ssl=True, min_size=5, max_size=20)
# 测试连接池
from databases import Database
import multiprocessing as mp
import config
database = Database(config.DATABASE_URL, min_size=2, max_size=5)
await database.connect()
def sleep():
print("hw")
# res = database.fetch_one("select sleep(1)")
print(res)
pool = mp.Pool(10)
for i in range(10):
pool.apply_async(sleep, ())
await database.disconnect()
await database.disconnect()