In [1]:
# 您可以通过将数据库用作异步上下文管理器来控制数据库的连接/断开连接。
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)
mysql://root:wxnacy@127.0.0.1:3306/study?charset=utf8mb4
(1, 'Daisy', 92)
In [ ]:
'''
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()
In [ ]:
# 连接选项
# Use an SSL connection.
database = Database('postgresql://localhost/example?ssl=true')
In [ ]:
# Use a connection pool of between 5-20 connections.
database = Database('mysql://localhost/example?min_size=5&max_size=20')
In [ ]:
# 使用键值对设置选项
database = Database('postgresql://localhost/example', ssl=True, min_size=5, max_size=20)
In [19]:
# 测试连接池

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()
In [16]:
await database.disconnect()
---------------------------------------------------------------------------
AssertionError                            Traceback (most recent call last)
<ipython-input-16-d0df956c05ea> in async-def-wrapper()

~/.pyenv/versions/3.7.6/Python.framework/Versions/3.7/lib/python3.7/site-packages/databases/core.py in disconnect(self)
    103         Close all connections in the connection pool.
    104         """
--> 105         assert self.is_connected, "Already disconnected."
    106 
    107         if self._force_rollback:

AssertionError: Already disconnected.
In [ ]: