# FastAPI DB
用于 fastapi 应用，可随时获取上下文，使用sqlalchemy，封装Base让模型类直接查询。
目前仅支持单数据库的应用，已封装CRUD模型，大幅度解放双手
## 对象说明
**FastAPIDBMiddleware**: 中间件通过add_middleware绑定，但需要运行到web才能触发
**FastAPIDB**：扩展应用，你可以传递FastAPI实例自动绑定事务中间件，不需要运行到web即可触发使用
**TransactionManager**: 事务管理器，用于创建会话，进入上下文，退出上下文等操作
**TransactionContext**: 事务上下文，sqlalchemy到会话也在里面，存放于ContextVar中
**local_transaction**： 局部上下文，你可以通过with创建局部事务上下文
**transactional**： 函数声明式事务，你可以使用它进行复杂的业务，而无需依次传递会话
**ctx**： 上下文代理器，你可以通过它随时访问session，但你要确保代码已经提前开启了事务
**CRUDModel**：增删改查模型，你一般情况不需要继承它，而是继承Model
## 安装
```shell
pip install fastapi_db
```
## 示例代码
```python
import random
import uvicorn
from fastapi import FastAPI
from fastapi_db import FastAPIDBMiddleware, Model, ctx
from sqlalchemy import Column, String, create_engine, Integer
app = FastAPI()
engine = create_engine('sqlite:///test.db')
app.add_middleware(FastAPIDBMiddleware, engine=engine)
class Base(Model):
__abstract__ = True
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(32), nullable=False, unique=True)
Base.metadata.create_all(engine)
@app.get('/user')
def get_list():
users = User.query().all()
return users
@app.get('/create')
def create_user(username: str = None):
if username is None:
username = str(random.randint(1, 99999))
user = User(username=username)
# 插入数据 会自动flush让数据库返回自增ID，你也可以通过flush参数控制是否返回自增ID
user.insert()
print(user.id)
# > 1
return user
if __name__ == '__main__':
uvicorn.run(app, host='0.0.0.0', port=9001)
```
## 如果你想在启动时查询数据
你需要注意的是每次local_transaction生产的事务都是不同的，你可以改变传播方式达到你想要的目的
```python
from fastapi import FastAPI
from fastapi_db import FastAPIDB, Model, ctx, local_transaction
from sqlalchemy import Column, String, create_engine, Integer
app = FastAPI()
engine = create_engine('sqlite:///test.db')
extension = FastAPIDB(app, engine=engine)
class Base(Model):
__abstract__ = True
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(32), nullable=False, unique=True)
def __repr__(self):
return f"User(id={self.id}, username={self.username})"
Base.metadata.create_all(engine)
with local_transaction() as context:
"""你会得到是个事务上下文对象, sqlalchemy会话也在其中"""
print(context)
# TransactionContext(session=<sqlalchemy.orm.session.Session object at 0x1209ea690>,
# propagation=Propagation.NEW, isolation=Isolation.DEFAULT, autocommit=True)
"""使用会话查询对象"""
users = context.session.query(User).all()
print(users)
# > [User(id=1, username=asd), User(id=2, username=6891), User(id=3, username=55896), User(id=4, username=74272),
# User(id=5, username=96389), User(id=6, username=91287), User(id=7, username=52207)]
"""使用代理器查询对象"""
users = ctx.session.query(User).all()
print(users)
# > [User(id=1, username=asd), User(id=2, username=6891), User(id=3, username=55896), User(id=4, username=74272),
# User(id=5, username=96389), User(id=6, username=91287), User(id=7, username=52207)]
"""使用模型查询"""
users = User.select_all()
print(users)
# > [User(id=1, username=asd), User(id=2, username=6891), User(id=3, username=55896), User(id=4, username=74272),
# User(id=5, username=96389), User(id=6, username=91287), User(id=7, username=52207)]
```
## 接口使用事务
默认情况下请求会自动开启事务，你可以使用transactional声明式事务覆盖请求事务，可以改变传播方式，自动提交，隔离级别，异常处理等
```python
import uvicorn
from fastapi import FastAPI
from sqlalchemy import Column, String, create_engine, Integer
from fastapi_db import Model, FastAPIDB, transactional
app = FastAPI()
engine = create_engine('sqlite:///test.db')
FastAPIDB(app, engine=engine)
class Base(Model):
__abstract__ = True
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(32), nullable=False, unique=True)
def __repr__(self):
return f"User(id={self.id}, username={self.username})"
Base.metadata.create_all(engine)
@app.get('/user')
@transactional(autocommit=False)
def get_list():
# 你可以禁止该接口自动提交，隔离级别或其它参数
users = User.query().all()
return users
if __name__ == '__main__':
uvicorn.run(app, host='0.0.0.0', port=9000)
```
## 声明式事务
你需要注意的是声明式事务和其它开启事务方式的区别，声明式事务默认传播方式为查找上级事务，如果不存在则创建。声明式事务一旦创建则不会自动关闭。
```python
from fastapi import FastAPI
from sqlalchemy import Column, String, create_engine, Integer
from fastapi_db import Model, FastAPIDB, ctx, transactional, Propagation, local_transaction
app = FastAPI()
engine = create_engine('sqlite:///test.db')
FastAPIDB(app, engine=engine)
class Base(Model):
__abstract__ = True
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(32), nullable=False, unique=True)
def __repr__(self):
return f"User(id={self.id}, username={self.username})"
Base.metadata.create_all(engine)
@transactional()
def test1():
"""transactional默认会使用父级会话，没有则创建"""
print(ctx.session)
# > <sqlalchemy.orm.session.Session object at 0x105a1ab50>
@transactional()
def test2():
"""transactional默认会使用父级会话，没有则创建，可以看到和test1的会话是同一个"""
print(ctx.session)
# > <sqlalchemy.orm.session.Session object at 0x105a1ab50>
@transactional(propagation=Propagation.NEW)
def test3():
# 使用传播方式修改事务开启方式，这里无论上级是否存在事务都会创建
print(ctx.session)
# > <sqlalchemy.orm.session.Session object at 0x110727d90>
test1()
test2()
test3()
```
## 异常和回滚捕捉
在默认情况下，如果使用事务过程中发生异常会导致事务自动回滚，你可能需要在回滚前后做点事情，
rollback_callback参数是回滚后会调用的方法，exception_callback是回滚前调用的方法，你可以返回False禁止自动回滚
```python
from sqlalchemy import Column, String, Integer
from fastapi_db import transactional, Model
class Base(Model):
__abstract__ = True
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(32), nullable=False, unique=True)
def __repr__(self):
return f"User(id={self.id}, username={self.username})"
def handle_rollback_error(err):
# 在这里处理事务回滚后的操作
print('异常实例', err)
# > test rollback
@transactional(rollback_callback=handle_rollback_error)
def service2():
User.delete_by_id(1)
raise RuntimeError('test rollback')
service2()
```
## 增删改查
本示例使用User作为模型。涉及ID等字段需要模型设置好主键。
如果你想限制字段查询，一般通过_columns参数
如果你想使用order_by查询，一般通过_order_by参数
如果你想限制数量查询，一般通过_limit参数
如果你想使用offset查询，一般通过_offset参数
```python
from sqlalchemy import Column, String, Integer
from fastapi_db import Model, ctx, Page
class Base(Model):
__abstract__ = True
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(32), nullable=False, unique=True)
def __repr__(self):
return f"User(id={self.id}, username={self.username})"
"""查询操作"""
# 通过ID查询一条记录
User.get_by_id(1)
# 通过表达式条件查询一条记录，等同 select_one，get_one
User.get(id=1)
# 也可以 User.get(User.id == 1)
# 通过表达式条件查询所有记录，可以限制数量，偏移和排序查询
User.select(User.id.in_([1, 2, 3, 4]), _order_by=User.id.desc(), _limit=10, _offset=0)
# 不通过任何条件查询所有记录
User.select_all()
# 通过表达式条件查询并翻页
User.select_page(Page(1, 20), id=1)
# 通过表达式条件查询并翻页和返回总记录数
count, users = User.select_page_with_count(Page(1, 20), id=1)
# 通过ID列表查询所有记录
User.select_batch_ids([1, 2, 3, 4])
"""写入操作"""
# 插入数据
user = User(username='fastapi_db')
user.insert()
# 注意insert操作会导致数据库flush返回自增ID，如果你不需要，可以通过flush参数控制
# 保存数据
user = User(id=1, username='fastapi_db')
user.save()
# save操作会根据主键查询对象的数据并决定是插入还是更新
# 更新操作
# id为1的数据将被更新
User.update(id=1, values={
User.username: 'app'
})
"""删除"""
# 删除，但是多了查询，你可以直接delete_by_id
user = User.get_by_id(1)
user.delete()
# 通过ID删除，删除成功会返回1
User.delete_by_id(1)
# 通过ID列表删除
User.delete_batch_ids([1, 2, 3, 4, 5])
# 通过表达式删除
User.delete_by_expressions(id=1)
"""基本操作"""
# 数据脱敏（不会再修改到数据库）
user = User.get_by_id(1)
user.expunge()
user.username = '****'
```
## 数据脱敏
```python
# 数据脱敏（不会再修改到数据库）
user = User.get_by_id(1)
user.expunge()
user.username = '****'
```
Raw data
{
"_id": null,
"home_page": "https://github.com/breezechime/fastapi_db",
"name": "fastapi-db",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.6",
"maintainer_email": null,
"keywords": "fastapi, db, sqlalchemy",
"author": "\u4e14\u542c\u98ce\u94c3\u3001\u6211\u662f\u6307\u9488*\u3001ZDLAY\u3001KYZ",
"author_email": "breezechime@163.com",
"download_url": "https://files.pythonhosted.org/packages/65/27/119e70e7088669dc9d3679fab2bd8be966c9732ec5ea537bc8f10479b017/fastapi_db-0.7.1.tar.gz",
"platform": null,
"description": "# FastAPI DB\n\n\u7528\u4e8e fastapi \u5e94\u7528\uff0c\u53ef\u968f\u65f6\u83b7\u53d6\u4e0a\u4e0b\u6587\uff0c\u4f7f\u7528sqlalchemy\uff0c\u5c01\u88c5Base\u8ba9\u6a21\u578b\u7c7b\u76f4\u63a5\u67e5\u8be2\u3002\n\u76ee\u524d\u4ec5\u652f\u6301\u5355\u6570\u636e\u5e93\u7684\u5e94\u7528\uff0c\u5df2\u5c01\u88c5CRUD\u6a21\u578b\uff0c\u5927\u5e45\u5ea6\u89e3\u653e\u53cc\u624b\n\n## \u5bf9\u8c61\u8bf4\u660e\n\n**FastAPIDBMiddleware**: \u4e2d\u95f4\u4ef6\u901a\u8fc7add_middleware\u7ed1\u5b9a\uff0c\u4f46\u9700\u8981\u8fd0\u884c\u5230web\u624d\u80fd\u89e6\u53d1\n\n**FastAPIDB**\uff1a\u6269\u5c55\u5e94\u7528\uff0c\u4f60\u53ef\u4ee5\u4f20\u9012FastAPI\u5b9e\u4f8b\u81ea\u52a8\u7ed1\u5b9a\u4e8b\u52a1\u4e2d\u95f4\u4ef6\uff0c\u4e0d\u9700\u8981\u8fd0\u884c\u5230web\u5373\u53ef\u89e6\u53d1\u4f7f\u7528\n\n**TransactionManager**: \u4e8b\u52a1\u7ba1\u7406\u5668\uff0c\u7528\u4e8e\u521b\u5efa\u4f1a\u8bdd\uff0c\u8fdb\u5165\u4e0a\u4e0b\u6587\uff0c\u9000\u51fa\u4e0a\u4e0b\u6587\u7b49\u64cd\u4f5c\n\n**TransactionContext**: \u4e8b\u52a1\u4e0a\u4e0b\u6587\uff0csqlalchemy\u5230\u4f1a\u8bdd\u4e5f\u5728\u91cc\u9762\uff0c\u5b58\u653e\u4e8eContextVar\u4e2d\n\n**local_transaction**\uff1a \u5c40\u90e8\u4e0a\u4e0b\u6587\uff0c\u4f60\u53ef\u4ee5\u901a\u8fc7with\u521b\u5efa\u5c40\u90e8\u4e8b\u52a1\u4e0a\u4e0b\u6587\n\n**transactional**\uff1a \u51fd\u6570\u58f0\u660e\u5f0f\u4e8b\u52a1\uff0c\u4f60\u53ef\u4ee5\u4f7f\u7528\u5b83\u8fdb\u884c\u590d\u6742\u7684\u4e1a\u52a1\uff0c\u800c\u65e0\u9700\u4f9d\u6b21\u4f20\u9012\u4f1a\u8bdd\n\n**ctx**\uff1a \u4e0a\u4e0b\u6587\u4ee3\u7406\u5668\uff0c\u4f60\u53ef\u4ee5\u901a\u8fc7\u5b83\u968f\u65f6\u8bbf\u95eesession\uff0c\u4f46\u4f60\u8981\u786e\u4fdd\u4ee3\u7801\u5df2\u7ecf\u63d0\u524d\u5f00\u542f\u4e86\u4e8b\u52a1\n\n**CRUDModel**\uff1a\u589e\u5220\u6539\u67e5\u6a21\u578b\uff0c\u4f60\u4e00\u822c\u60c5\u51b5\u4e0d\u9700\u8981\u7ee7\u627f\u5b83\uff0c\u800c\u662f\u7ee7\u627fModel\n\n## \u5b89\u88c5\n\n```shell\npip install fastapi_db\n```\n\n## \u793a\u4f8b\u4ee3\u7801\n\n```python\nimport random\nimport uvicorn\nfrom fastapi import FastAPI\nfrom fastapi_db import FastAPIDBMiddleware, Model, ctx\nfrom sqlalchemy import Column, String, create_engine, Integer\n\napp = FastAPI()\n\nengine = create_engine('sqlite:///test.db')\napp.add_middleware(FastAPIDBMiddleware, engine=engine)\n\n\nclass Base(Model):\n __abstract__ = True\n\n\nclass User(Base):\n __tablename__ = 'user'\n\n id = Column(Integer, primary_key=True, autoincrement=True)\n username = Column(String(32), nullable=False, unique=True)\n\n\nBase.metadata.create_all(engine)\n\n\n@app.get('/user')\ndef get_list():\n users = User.query().all()\n return users\n\n\n@app.get('/create')\ndef create_user(username: str = None):\n if username is None:\n username = str(random.randint(1, 99999))\n user = User(username=username)\n # \u63d2\u5165\u6570\u636e \u4f1a\u81ea\u52a8flush\u8ba9\u6570\u636e\u5e93\u8fd4\u56de\u81ea\u589eID\uff0c\u4f60\u4e5f\u53ef\u4ee5\u901a\u8fc7flush\u53c2\u6570\u63a7\u5236\u662f\u5426\u8fd4\u56de\u81ea\u589eID\n user.insert()\n print(user.id)\n # > 1\n return user\n\n\nif __name__ == '__main__':\n uvicorn.run(app, host='0.0.0.0', port=9001)\n```\n\n\n## \u5982\u679c\u4f60\u60f3\u5728\u542f\u52a8\u65f6\u67e5\u8be2\u6570\u636e\n\n\u4f60\u9700\u8981\u6ce8\u610f\u7684\u662f\u6bcf\u6b21local_transaction\u751f\u4ea7\u7684\u4e8b\u52a1\u90fd\u662f\u4e0d\u540c\u7684\uff0c\u4f60\u53ef\u4ee5\u6539\u53d8\u4f20\u64ad\u65b9\u5f0f\u8fbe\u5230\u4f60\u60f3\u8981\u7684\u76ee\u7684\n\n```python\nfrom fastapi import FastAPI\nfrom fastapi_db import FastAPIDB, Model, ctx, local_transaction\nfrom sqlalchemy import Column, String, create_engine, Integer\n\napp = FastAPI()\n\nengine = create_engine('sqlite:///test.db')\nextension = FastAPIDB(app, engine=engine)\n\n\nclass Base(Model):\n __abstract__ = True\n\n\nclass User(Base):\n __tablename__ = 'user'\n\n id = Column(Integer, primary_key=True, autoincrement=True)\n username = Column(String(32), nullable=False, unique=True)\n \n def __repr__(self):\n return f\"User(id={self.id}, username={self.username})\"\n\nBase.metadata.create_all(engine)\n\nwith local_transaction() as context:\n\n \"\"\"\u4f60\u4f1a\u5f97\u5230\u662f\u4e2a\u4e8b\u52a1\u4e0a\u4e0b\u6587\u5bf9\u8c61, sqlalchemy\u4f1a\u8bdd\u4e5f\u5728\u5176\u4e2d\"\"\"\n print(context)\n # TransactionContext(session=<sqlalchemy.orm.session.Session object at 0x1209ea690>,\n # propagation=Propagation.NEW, isolation=Isolation.DEFAULT, autocommit=True)\n\n \"\"\"\u4f7f\u7528\u4f1a\u8bdd\u67e5\u8be2\u5bf9\u8c61\"\"\"\n users = context.session.query(User).all()\n print(users)\n # > [User(id=1, username=asd), User(id=2, username=6891), User(id=3, username=55896), User(id=4, username=74272),\n # User(id=5, username=96389), User(id=6, username=91287), User(id=7, username=52207)]\n\n \"\"\"\u4f7f\u7528\u4ee3\u7406\u5668\u67e5\u8be2\u5bf9\u8c61\"\"\"\n users = ctx.session.query(User).all()\n print(users)\n # > [User(id=1, username=asd), User(id=2, username=6891), User(id=3, username=55896), User(id=4, username=74272),\n # User(id=5, username=96389), User(id=6, username=91287), User(id=7, username=52207)]\n\n \"\"\"\u4f7f\u7528\u6a21\u578b\u67e5\u8be2\"\"\"\n users = User.select_all()\n print(users)\n # > [User(id=1, username=asd), User(id=2, username=6891), User(id=3, username=55896), User(id=4, username=74272),\n # User(id=5, username=96389), User(id=6, username=91287), User(id=7, username=52207)]\n```\n\n## \u63a5\u53e3\u4f7f\u7528\u4e8b\u52a1\n\n\u9ed8\u8ba4\u60c5\u51b5\u4e0b\u8bf7\u6c42\u4f1a\u81ea\u52a8\u5f00\u542f\u4e8b\u52a1\uff0c\u4f60\u53ef\u4ee5\u4f7f\u7528transactional\u58f0\u660e\u5f0f\u4e8b\u52a1\u8986\u76d6\u8bf7\u6c42\u4e8b\u52a1\uff0c\u53ef\u4ee5\u6539\u53d8\u4f20\u64ad\u65b9\u5f0f\uff0c\u81ea\u52a8\u63d0\u4ea4\uff0c\u9694\u79bb\u7ea7\u522b\uff0c\u5f02\u5e38\u5904\u7406\u7b49\n\n```python\nimport uvicorn\nfrom fastapi import FastAPI\nfrom sqlalchemy import Column, String, create_engine, Integer\n\nfrom fastapi_db import Model, FastAPIDB, transactional\n\napp = FastAPI()\n\nengine = create_engine('sqlite:///test.db')\nFastAPIDB(app, engine=engine)\n\n\nclass Base(Model):\n __abstract__ = True\n\n\nclass User(Base):\n __tablename__ = 'user'\n\n id = Column(Integer, primary_key=True, autoincrement=True)\n username = Column(String(32), nullable=False, unique=True)\n\n def __repr__(self):\n return f\"User(id={self.id}, username={self.username})\"\n\n\nBase.metadata.create_all(engine)\n\n\n@app.get('/user')\n@transactional(autocommit=False)\ndef get_list():\n # \u4f60\u53ef\u4ee5\u7981\u6b62\u8be5\u63a5\u53e3\u81ea\u52a8\u63d0\u4ea4\uff0c\u9694\u79bb\u7ea7\u522b\u6216\u5176\u5b83\u53c2\u6570\n users = User.query().all()\n return users\n\n\nif __name__ == '__main__':\n uvicorn.run(app, host='0.0.0.0', port=9000)\n```\n\n## \u58f0\u660e\u5f0f\u4e8b\u52a1\n\n\u4f60\u9700\u8981\u6ce8\u610f\u7684\u662f\u58f0\u660e\u5f0f\u4e8b\u52a1\u548c\u5176\u5b83\u5f00\u542f\u4e8b\u52a1\u65b9\u5f0f\u7684\u533a\u522b\uff0c\u58f0\u660e\u5f0f\u4e8b\u52a1\u9ed8\u8ba4\u4f20\u64ad\u65b9\u5f0f\u4e3a\u67e5\u627e\u4e0a\u7ea7\u4e8b\u52a1\uff0c\u5982\u679c\u4e0d\u5b58\u5728\u5219\u521b\u5efa\u3002\u58f0\u660e\u5f0f\u4e8b\u52a1\u4e00\u65e6\u521b\u5efa\u5219\u4e0d\u4f1a\u81ea\u52a8\u5173\u95ed\u3002\n\n```python\nfrom fastapi import FastAPI\nfrom sqlalchemy import Column, String, create_engine, Integer\nfrom fastapi_db import Model, FastAPIDB, ctx, transactional, Propagation, local_transaction\n\napp = FastAPI()\n\nengine = create_engine('sqlite:///test.db')\nFastAPIDB(app, engine=engine)\n\n\nclass Base(Model):\n __abstract__ = True\n\n\nclass User(Base):\n __tablename__ = 'user'\n\n id = Column(Integer, primary_key=True, autoincrement=True)\n username = Column(String(32), nullable=False, unique=True)\n\n def __repr__(self):\n return f\"User(id={self.id}, username={self.username})\"\n\n\nBase.metadata.create_all(engine)\n\n\n@transactional()\ndef test1():\n \"\"\"transactional\u9ed8\u8ba4\u4f1a\u4f7f\u7528\u7236\u7ea7\u4f1a\u8bdd\uff0c\u6ca1\u6709\u5219\u521b\u5efa\"\"\"\n print(ctx.session)\n # > <sqlalchemy.orm.session.Session object at 0x105a1ab50>\n\n\n@transactional()\ndef test2():\n \"\"\"transactional\u9ed8\u8ba4\u4f1a\u4f7f\u7528\u7236\u7ea7\u4f1a\u8bdd\uff0c\u6ca1\u6709\u5219\u521b\u5efa\uff0c\u53ef\u4ee5\u770b\u5230\u548ctest1\u7684\u4f1a\u8bdd\u662f\u540c\u4e00\u4e2a\"\"\"\n print(ctx.session)\n # > <sqlalchemy.orm.session.Session object at 0x105a1ab50>\n\n\n@transactional(propagation=Propagation.NEW)\ndef test3():\n # \u4f7f\u7528\u4f20\u64ad\u65b9\u5f0f\u4fee\u6539\u4e8b\u52a1\u5f00\u542f\u65b9\u5f0f\uff0c\u8fd9\u91cc\u65e0\u8bba\u4e0a\u7ea7\u662f\u5426\u5b58\u5728\u4e8b\u52a1\u90fd\u4f1a\u521b\u5efa\n print(ctx.session)\n # > <sqlalchemy.orm.session.Session object at 0x110727d90>\n\n\ntest1()\ntest2()\ntest3()\n```\n\n## \u5f02\u5e38\u548c\u56de\u6eda\u6355\u6349\n\n\u5728\u9ed8\u8ba4\u60c5\u51b5\u4e0b\uff0c\u5982\u679c\u4f7f\u7528\u4e8b\u52a1\u8fc7\u7a0b\u4e2d\u53d1\u751f\u5f02\u5e38\u4f1a\u5bfc\u81f4\u4e8b\u52a1\u81ea\u52a8\u56de\u6eda\uff0c\u4f60\u53ef\u80fd\u9700\u8981\u5728\u56de\u6eda\u524d\u540e\u505a\u70b9\u4e8b\u60c5\uff0c\nrollback_callback\u53c2\u6570\u662f\u56de\u6eda\u540e\u4f1a\u8c03\u7528\u7684\u65b9\u6cd5\uff0cexception_callback\u662f\u56de\u6eda\u524d\u8c03\u7528\u7684\u65b9\u6cd5\uff0c\u4f60\u53ef\u4ee5\u8fd4\u56deFalse\u7981\u6b62\u81ea\u52a8\u56de\u6eda\n\n```python\nfrom sqlalchemy import Column, String, Integer\nfrom fastapi_db import transactional, Model\n\nclass Base(Model):\n __abstract__ = True\n\n\nclass User(Base):\n __tablename__ = 'user'\n\n id = Column(Integer, primary_key=True, autoincrement=True)\n username = Column(String(32), nullable=False, unique=True)\n\n def __repr__(self):\n return f\"User(id={self.id}, username={self.username})\"\n \ndef handle_rollback_error(err):\n # \u5728\u8fd9\u91cc\u5904\u7406\u4e8b\u52a1\u56de\u6eda\u540e\u7684\u64cd\u4f5c\n print('\u5f02\u5e38\u5b9e\u4f8b', err)\n # > test rollback\n\n\n@transactional(rollback_callback=handle_rollback_error)\ndef service2():\n User.delete_by_id(1)\n\n raise RuntimeError('test rollback')\n\n\nservice2()\n```\n\n\n## \u589e\u5220\u6539\u67e5\n\n\u672c\u793a\u4f8b\u4f7f\u7528User\u4f5c\u4e3a\u6a21\u578b\u3002\u6d89\u53caID\u7b49\u5b57\u6bb5\u9700\u8981\u6a21\u578b\u8bbe\u7f6e\u597d\u4e3b\u952e\u3002\n\n\u5982\u679c\u4f60\u60f3\u9650\u5236\u5b57\u6bb5\u67e5\u8be2\uff0c\u4e00\u822c\u901a\u8fc7_columns\u53c2\u6570\n\u5982\u679c\u4f60\u60f3\u4f7f\u7528order_by\u67e5\u8be2\uff0c\u4e00\u822c\u901a\u8fc7_order_by\u53c2\u6570\n\u5982\u679c\u4f60\u60f3\u9650\u5236\u6570\u91cf\u67e5\u8be2\uff0c\u4e00\u822c\u901a\u8fc7_limit\u53c2\u6570\n\u5982\u679c\u4f60\u60f3\u4f7f\u7528offset\u67e5\u8be2\uff0c\u4e00\u822c\u901a\u8fc7_offset\u53c2\u6570\n\n```python\nfrom sqlalchemy import Column, String, Integer\nfrom fastapi_db import Model, ctx, Page\n\nclass Base(Model):\n __abstract__ = True\n\n\nclass User(Base):\n __tablename__ = 'user'\n\n id = Column(Integer, primary_key=True, autoincrement=True)\n username = Column(String(32), nullable=False, unique=True)\n\n def __repr__(self):\n return f\"User(id={self.id}, username={self.username})\"\n\n\"\"\"\u67e5\u8be2\u64cd\u4f5c\"\"\"\n\n# \u901a\u8fc7ID\u67e5\u8be2\u4e00\u6761\u8bb0\u5f55\nUser.get_by_id(1)\n# \u901a\u8fc7\u8868\u8fbe\u5f0f\u6761\u4ef6\u67e5\u8be2\u4e00\u6761\u8bb0\u5f55\uff0c\u7b49\u540c select_one\uff0cget_one\nUser.get(id=1)\n# \u4e5f\u53ef\u4ee5 User.get(User.id == 1)\n\n# \u901a\u8fc7\u8868\u8fbe\u5f0f\u6761\u4ef6\u67e5\u8be2\u6240\u6709\u8bb0\u5f55\uff0c\u53ef\u4ee5\u9650\u5236\u6570\u91cf\uff0c\u504f\u79fb\u548c\u6392\u5e8f\u67e5\u8be2\nUser.select(User.id.in_([1, 2, 3, 4]), _order_by=User.id.desc(), _limit=10, _offset=0)\n# \u4e0d\u901a\u8fc7\u4efb\u4f55\u6761\u4ef6\u67e5\u8be2\u6240\u6709\u8bb0\u5f55\nUser.select_all()\n# \u901a\u8fc7\u8868\u8fbe\u5f0f\u6761\u4ef6\u67e5\u8be2\u5e76\u7ffb\u9875\nUser.select_page(Page(1, 20), id=1)\n# \u901a\u8fc7\u8868\u8fbe\u5f0f\u6761\u4ef6\u67e5\u8be2\u5e76\u7ffb\u9875\u548c\u8fd4\u56de\u603b\u8bb0\u5f55\u6570\ncount, users = User.select_page_with_count(Page(1, 20), id=1)\n# \u901a\u8fc7ID\u5217\u8868\u67e5\u8be2\u6240\u6709\u8bb0\u5f55\nUser.select_batch_ids([1, 2, 3, 4])\n\n\n\"\"\"\u5199\u5165\u64cd\u4f5c\"\"\"\n\n# \u63d2\u5165\u6570\u636e\nuser = User(username='fastapi_db')\nuser.insert()\n\n# \u6ce8\u610finsert\u64cd\u4f5c\u4f1a\u5bfc\u81f4\u6570\u636e\u5e93flush\u8fd4\u56de\u81ea\u589eID\uff0c\u5982\u679c\u4f60\u4e0d\u9700\u8981\uff0c\u53ef\u4ee5\u901a\u8fc7flush\u53c2\u6570\u63a7\u5236\n\n# \u4fdd\u5b58\u6570\u636e\n\nuser = User(id=1, username='fastapi_db')\nuser.save()\n\n# save\u64cd\u4f5c\u4f1a\u6839\u636e\u4e3b\u952e\u67e5\u8be2\u5bf9\u8c61\u7684\u6570\u636e\u5e76\u51b3\u5b9a\u662f\u63d2\u5165\u8fd8\u662f\u66f4\u65b0\n\n# \u66f4\u65b0\u64cd\u4f5c\n\n# id\u4e3a1\u7684\u6570\u636e\u5c06\u88ab\u66f4\u65b0\nUser.update(id=1, values={\n User.username: 'app'\n})\n\n\"\"\"\u5220\u9664\"\"\"\n\n\n# \u5220\u9664\uff0c\u4f46\u662f\u591a\u4e86\u67e5\u8be2\uff0c\u4f60\u53ef\u4ee5\u76f4\u63a5delete_by_id\nuser = User.get_by_id(1)\nuser.delete()\n\n# \u901a\u8fc7ID\u5220\u9664\uff0c\u5220\u9664\u6210\u529f\u4f1a\u8fd4\u56de1\nUser.delete_by_id(1)\n\n# \u901a\u8fc7ID\u5217\u8868\u5220\u9664\nUser.delete_batch_ids([1, 2, 3, 4, 5])\n\n# \u901a\u8fc7\u8868\u8fbe\u5f0f\u5220\u9664\nUser.delete_by_expressions(id=1)\n\n\"\"\"\u57fa\u672c\u64cd\u4f5c\"\"\"\n\n# \u6570\u636e\u8131\u654f\uff08\u4e0d\u4f1a\u518d\u4fee\u6539\u5230\u6570\u636e\u5e93\uff09\nuser = User.get_by_id(1)\nuser.expunge()\nuser.username = '****'\n\n\n```\n\n## \u6570\u636e\u8131\u654f\n\n```python\n# \u6570\u636e\u8131\u654f\uff08\u4e0d\u4f1a\u518d\u4fee\u6539\u5230\u6570\u636e\u5e93\uff09\nuser = User.get_by_id(1)\nuser.expunge()\nuser.username = '****'\n```\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "fastapi\u4f7f\u7528sqlalchemy\u4efb\u610f\u63a5\u53e3\u4f4d\u7f6e\u83b7\u53d6db\u4e0a\u4e0b\u6587\uff0c\u975e\u5e38\u5b89\u5168\u5e76\u53ef\u9760\u3002",
"version": "0.7.1",
"project_urls": {
"Homepage": "https://github.com/breezechime/fastapi_db"
},
"split_keywords": [
"fastapi",
" db",
" sqlalchemy"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "bcfe3c63d1531a1332084cee75f8734c1ff0ca89d3916161af3a0190faabb0fd",
"md5": "98d8a3924bd2157fb9715666e967377d",
"sha256": "b88aefa1a94633b8afab47438d566cf0bd72264b27ef33803d8a059d4034df82"
},
"downloads": -1,
"filename": "fastapi_db-0.7.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "98d8a3924bd2157fb9715666e967377d",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.6",
"size": 17050,
"upload_time": "2024-07-31T07:34:43",
"upload_time_iso_8601": "2024-07-31T07:34:43.159990Z",
"url": "https://files.pythonhosted.org/packages/bc/fe/3c63d1531a1332084cee75f8734c1ff0ca89d3916161af3a0190faabb0fd/fastapi_db-0.7.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "6527119e70e7088669dc9d3679fab2bd8be966c9732ec5ea537bc8f10479b017",
"md5": "ec24fef9753d3940a86cb547506a7182",
"sha256": "afd0c7fcc2524a37aec48f7f5cc2ecea082f0756c7811266e6eb26631f820d56"
},
"downloads": -1,
"filename": "fastapi_db-0.7.1.tar.gz",
"has_sig": false,
"md5_digest": "ec24fef9753d3940a86cb547506a7182",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.6",
"size": 17245,
"upload_time": "2024-07-31T07:34:46",
"upload_time_iso_8601": "2024-07-31T07:34:46.295580Z",
"url": "https://files.pythonhosted.org/packages/65/27/119e70e7088669dc9d3679fab2bd8be966c9732ec5ea537bc8f10479b017/fastapi_db-0.7.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-07-31 07:34:46",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "breezechime",
"github_project": "fastapi_db",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "fastapi-db"
}
JSON
