pip install databases
pip install databases[aiomysql]
pip install pymysql

from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

DATABASE_URL = "mysql+pymysql://root:chenhaifeng@47.120.55.99:3306/fastapi"

# 实例化数据库连接
database = databases.Database(DATABASE_URL)

# 实例化数据库对象
metadata = sqlalchemy.MetaData()

# 创建表
notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String(56)),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


# 创建一个发动机
engine = sqlalchemy.create_engine(
    DATABASE_URL
)

# 表迁移
metadata.create_all(engine)


# 创建模型
class NoteIn(BaseModel):
    text: str
    completed: bool


# 创建模型
class Note(BaseModel):
    id: int
    text: str
    completed: bool

    class Config:
        orm_mode = True


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

from databases import Database

database = Database('postgresql+asyncpg://localhost/example')


# Establish the connection pool
await database.connect()

# Execute
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query=query, values=values)

# Execute many
query = notes.insert()
values = [
    {"text": "example2", "completed": False},
    {"text": "example3", "completed": True},
]
await database.execute_many(query=query, values=values)

# Fetch multiple rows
query = notes.select()
rows = await database.fetch_all(query=query)

# Fetch single row
query = notes.select()
row = await database.fetch_one(query=query)

# Fetch single value, defaults to `column=0`.
query = notes.select()
value = await database.fetch_val(query=query)

# Fetch multiple rows without loading them all into memory at once
query = notes.select()
async for row in database.iterate(query=query):
    ...

# Close all connections in the connection pool
await database.disconnect()

query = "SELECT * FROM notes WHERE id = :id"
result = await database.fetch_one(query=query, values={"id": 1})
result.id  # Access field via attribute
result._mapping['id']  # Access field via mapping

一.
async with database.transaction():
    ...


二.
transaction = await database.transaction()
try:
    ...
except:
    await transaction.rollback()
else:
    await transaction.commit()