创建CRUD基类，简化操作。
This commit is contained in:
?..濡..
parent
9e989e3421
commit
9acac1711e
65
README.md
65
README.md
@ -258,6 +258,71 @@ from mooc.models.account import AccountWechats # 新增,将创建好的模型
|
||||
python main.py
|
||||
```
|
||||
|
||||
## CRUDBase 使用指南
|
||||
|
||||
### 简介
|
||||
|
||||
CRUDBase是一个通用的CRUD基类，提供了基础的增删改查操作。通过继承这个基类，可以快速实现特定模型的CRUD操作。
|
||||
|
||||
### 使用方法
|
||||
|
||||
1. 创建CRUD类:
|
||||
|
||||
```python
|
||||
from mooc.crud.base import CRUDBase
|
||||
from mooc.models.your_model import YourModel
|
||||
from mooc.schemas.your_schema import YourCreateSchema, YourUpdateSchema
|
||||
|
||||
class CRUDYourModel(CRUDBase[YourModel, YourCreateSchema, YourUpdateSchema]):
|
||||
pass # 继承基础功能即可，如需自定义方法可以在这里添加
|
||||
```
|
||||
|
||||
2. 实例化CRUD对象:
|
||||
|
||||
```python
|
||||
crud_your_model = CRUDYourModel(YourModel)
|
||||
```
|
||||
|
||||
### 基础操作示例
|
||||
|
||||
```python
|
||||
# 创建记录
|
||||
new_item = crud_your_model.create(db, obj_in=item_create_schema)
|
||||
|
||||
# 获取单条记录
|
||||
item = crud_your_model.get(db, id=123)
|
||||
|
||||
# 获取多条记录
|
||||
items = crud_your_model.get_multi(db, skip=0, limit=100)
|
||||
|
||||
# 更新记录
|
||||
updated_item = crud_your_model.update(db, db_obj=existing_item, obj_in=item_update_schema)
|
||||
|
||||
# 删除记录
|
||||
crud_your_model.remove(db, id=123)
|
||||
```
|
||||
|
||||
### 自定义方法示例
|
||||
|
||||
```python
|
||||
class CRUDYourModel(CRUDBase[YourModel, YourCreateSchema, YourUpdateSchema]):
|
||||
def get_by_custom_field(self, db: Session, field_value: str) -> Optional[YourModel]:
|
||||
return db.query(self.model).filter(self.model.custom_field == field_value).first()
|
||||
```
|
||||
|
||||
### 注意事项
|
||||
|
||||
1. CRUDBase需要三个类型参数：
|
||||
- ModelType: SQLAlchemy模型类
|
||||
- CreateSchemaType: Pydantic创建模型
|
||||
- UpdateSchemaType: Pydantic更新模型
|
||||
|
||||
2. 所有方法都需要传入数据库会话(db: Session)参数
|
||||
|
||||
3. 更新操作支持两种方式：
|
||||
- 字典形式：`update(db, db_obj=item, obj_in={"field": "new_value"})`
|
||||
- Pydantic模型：`update(db, db_obj=item, obj_in=ItemUpdate(field="new_value"))`
|
||||
|
||||
## API文档
|
||||
|
||||
启动服务后访问: http://localhost:2333/docs
|
||||
|
||||
@ -2,7 +2,7 @@ from typing import List
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from sqlalchemy.orm import Session
|
||||
from mooc.api import deps
|
||||
from mooc.crud.crud_account import ims_account_baiduapp
|
||||
from mooc.crud.crud_account import account_baiduapp
|
||||
from mooc.schemas.account import AccountBaiduappCreate, AccountBaiduappRead, AccountBaiduappUpdate
|
||||
|
||||
account_router = APIRouter()
|
||||
@ -16,7 +16,7 @@ def create_baiduapp(
|
||||
"""
|
||||
创建百度小程序账号
|
||||
"""
|
||||
baiduapp = ims_account_baiduapp.create(db=db, obj_in=baiduapp_in)
|
||||
baiduapp = account_baiduapp.create(db=db, obj_in=baiduapp_in)
|
||||
return baiduapp
|
||||
|
||||
@account_router.get("/baiduapp/{acid}", response_model=AccountBaiduappRead)
|
||||
@ -28,7 +28,7 @@ def read_baiduapp(
|
||||
"""
|
||||
获取百度小程序账号信息
|
||||
"""
|
||||
baiduapp = ims_account_baiduapp.get(db=db, acid=acid)
|
||||
baiduapp = account_baiduapp.get(db=db, acid=acid)
|
||||
if not baiduapp:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
@ -46,13 +46,13 @@ def update_baiduapp(
|
||||
"""
|
||||
更新百度小程序账号信息
|
||||
"""
|
||||
baiduapp = ims_account_baiduapp.get(db=db, acid=acid)
|
||||
baiduapp = account_baiduapp.get(db=db, acid=acid)
|
||||
if not baiduapp:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Baidu app not found"
|
||||
)
|
||||
baiduapp = ims_account_baiduapp.update(
|
||||
baiduapp = account_baiduapp.update(
|
||||
db=db,
|
||||
db_obj=baiduapp,
|
||||
obj_in=baiduapp_in
|
||||
@ -68,11 +68,11 @@ def delete_baiduapp(
|
||||
"""
|
||||
删除百度小程序账号
|
||||
"""
|
||||
baiduapp = ims_account_baiduapp.get(db=db, acid=acid)
|
||||
baiduapp = account_baiduapp.get(db=db, acid=acid)
|
||||
if not baiduapp:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Baidu app not found"
|
||||
)
|
||||
baiduapp = ims_account_baiduapp.delete(db=db, acid=acid)
|
||||
baiduapp = account_baiduapp.delete(db=db, acid=acid)
|
||||
return baiduapp
|
||||
@ -1,276 +1,65 @@
|
||||
from sqlalchemy.orm import Session
|
||||
from typing import Optional
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
from mooc.crud.crud_base import CRUDBase
|
||||
from mooc.models.account import (
|
||||
AccountWechats,
|
||||
AccountAliapp,
|
||||
AccountBaiduapp,
|
||||
AccountPhoneapp,
|
||||
AccountToutiaoapp,
|
||||
AccountWebapp,
|
||||
Account,
|
||||
AccountWxapp,
|
||||
AccountXzapp
|
||||
)
|
||||
AccountWechats, AccountAliapp, AccountBaiduapp,
|
||||
AccountPhoneapp, AccountToutiaoapp, Account,
|
||||
AccountWebapp, AccountWxapp, AccountXzapp
|
||||
)
|
||||
from mooc.schemas.account import (
|
||||
AccountWechatsCreate,
|
||||
AccountWechatsUpdate,
|
||||
AccountAliappCreate,
|
||||
AccountAliappUpdate,
|
||||
AccountBaiduappCreate,
|
||||
AccountBaiduappUpdate,
|
||||
AccountPhoneappCreate,
|
||||
AccountPhoneappUpdate,
|
||||
AccountToutiaoappCreate,
|
||||
AccountToutiaoappUpdate,
|
||||
AccountWebappCreate,
|
||||
AccountWebappUpdate,
|
||||
AccountXzappCreate,
|
||||
AccountXzappUpdate,
|
||||
AccountWxappCreate,
|
||||
AccountWxappUpdate
|
||||
|
||||
AccountWechatsCreate, AccountWechatsUpdate,
|
||||
AccountAliappCreate, AccountAliappUpdate,
|
||||
AccountBaiduappCreate, AccountBaiduappUpdate,
|
||||
AccountPhoneappCreate, AccountPhoneappUpdate,
|
||||
AccountToutiaoappCreate, AccountToutiaoappUpdate,
|
||||
AccountWebappCreate, AccountWebappUpdate,
|
||||
AccountWxappCreate, AccountWxappUpdate,
|
||||
AccountXzappCreate, AccountXzappUpdate
|
||||
)
|
||||
|
||||
class CRUDAccountWechats:
|
||||
def create(self, db: Session, obj_in: AccountWechatsCreate) -> AccountWechats:
|
||||
db_obj = AccountWechats(**obj_in.dict())
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
class CRUDAccountWechats(CRUDBase[AccountWechats, AccountWechatsCreate, AccountWechatsUpdate]):
|
||||
def get_by_uniacid(self, db: Session, *, uniacid: int) -> Optional[AccountWechats]:
|
||||
return db.query(self.model).filter(self.model.uniacid == uniacid).first()
|
||||
|
||||
def get(self, db: Session, acid: int) -> Optional[AccountWechats]:
|
||||
return db.query(AccountWechats).filter(AccountWechats.acid == acid).first()
|
||||
class CRUDAccountAliapp(CRUDBase[AccountAliapp, AccountAliappCreate, AccountAliappUpdate]):
|
||||
def get_by_uniacid(self, db: Session, *, uniacid: int) -> Optional[AccountAliapp]:
|
||||
return db.query(self.model).filter(self.model.uniacid == uniacid).first()
|
||||
|
||||
def update(
|
||||
self, db: Session, *, db_obj: AccountWechats, obj_in: AccountWechatsUpdate
|
||||
) -> AccountWechats:
|
||||
for field, value in obj_in.dict(exclude_unset=True).items():
|
||||
setattr(db_obj, field, value)
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
class CRUDAccountBaiduapp(CRUDBase[AccountBaiduapp, AccountBaiduappCreate, AccountBaiduappUpdate]):
|
||||
def get_by_uniacid(self, db: Session, *, uniacid: int) -> Optional[AccountBaiduapp]:
|
||||
return db.query(self.model).filter(self.model.uniacid == uniacid).first()
|
||||
|
||||
def delete(self, db: Session, acid: int) -> None:
|
||||
obj = db.query(AccountWechats).filter(AccountWechats.acid == acid).first()
|
||||
if obj:
|
||||
db.delete(obj)
|
||||
db.commit()
|
||||
class CRUDAccountPhoneapp(CRUDBase[AccountPhoneapp, AccountPhoneappCreate, AccountPhoneappUpdate]):
|
||||
def get_by_uniacid(self, db: Session, *, uniacid: int) -> Optional[AccountPhoneapp]:
|
||||
return db.query(self.model).filter(self.model.uniacid == uniacid).first()
|
||||
|
||||
class CRUDAccountToutiaoapp(CRUDBase[AccountToutiaoapp, AccountToutiaoappCreate, AccountToutiaoappUpdate]):
|
||||
def get_by_uniacid(self, db: Session, *, uniacid: int) -> Optional[AccountToutiaoapp]:
|
||||
return db.query(self.model).filter(self.model.uniacid == uniacid).first()
|
||||
|
||||
class CRUDAccountAliapp:
|
||||
def create(self, db: Session, obj_in: AccountAliappCreate) -> AccountAliapp:
|
||||
db_obj = AccountAliapp(**obj_in.dict())
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
class CRUDAccountWebapp(CRUDBase[AccountWebapp, AccountWebappCreate, AccountWebappUpdate]):
|
||||
def get_by_uniacid(self, db: Session, *, uniacid: int) -> Optional[AccountWebapp]:
|
||||
return db.query(self.model).filter(self.model.uniacid == uniacid).first()
|
||||
|
||||
def get(self, db: Session, acid: int) -> Optional[AccountAliapp]:
|
||||
return db.query(AccountAliapp).filter(AccountAliapp.acid == acid).first()
|
||||
class CRUDAccountWxapp(CRUDBase[AccountWxapp, AccountWxappCreate, AccountWxappUpdate]):
|
||||
def get_by_uniacid(self, db: Session, *, uniacid: int) -> Optional[AccountWxapp]:
|
||||
return db.query(self.model).filter(self.model.uniacid == uniacid).first()
|
||||
|
||||
def update(
|
||||
self, db: Session, *, db_obj: AccountAliapp, obj_in: AccountAliappUpdate
|
||||
) -> AccountAliapp:
|
||||
for field, value in obj_in.dict(exclude_unset=True).items():
|
||||
setattr(db_obj, field, value)
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
class CRUDAccountXzapp(CRUDBase[AccountXzapp, AccountXzappCreate, AccountXzappUpdate]):
|
||||
def get_by_uniacid(self, db: Session, *, uniacid: int) -> Optional[AccountXzapp]:
|
||||
return db.query(self.model).filter(self.model.uniacid == uniacid).first()
|
||||
|
||||
def delete(self, db: Session, acid: int) -> None:
|
||||
obj = db.query(AccountAliapp).filter(AccountAliapp.acid == acid).first()
|
||||
if obj:
|
||||
db.delete(obj)
|
||||
db.commit()
|
||||
class CRUDAccountBaiduapp:
|
||||
def create(self, db: Session, obj_in: AccountBaiduappCreate) -> AccountBaiduapp:
|
||||
db_obj = AccountBaiduapp(**obj_in.dict())
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def get(self, db: Session, acid: int) -> Optional[AccountBaiduapp]:
|
||||
return db.query(AccountBaiduapp).filter(AccountBaiduapp.acid == acid).first()
|
||||
|
||||
def update(
|
||||
self, db: Session, *, db_obj: AccountBaiduapp, obj_in: AccountBaiduappUpdate
|
||||
) -> AccountBaiduapp:
|
||||
for field, value in obj_in.dict(exclude_unset=True).items():
|
||||
setattr(db_obj, field, value)
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def delete(self, db: Session, acid: int) -> None:
|
||||
obj = db.query(AccountBaiduapp).filter(AccountBaiduapp.acid == acid).first()
|
||||
if obj:
|
||||
db.delete(obj)
|
||||
db.commit()
|
||||
class CRUDAccountPhoneapp:
|
||||
def create(self, db: Session, obj_in: AccountPhoneappCreate) -> AccountPhoneapp:
|
||||
db_obj = AccountPhoneapp(**obj_in.dict())
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def get(self, db: Session, acid: int) -> Optional[AccountPhoneapp]:
|
||||
return db.query(AccountPhoneapp).filter(AccountPhoneapp.acid == acid).first()
|
||||
|
||||
def update(
|
||||
self, db: Session, *, db_obj: AccountPhoneapp, obj_in: AccountPhoneappUpdate
|
||||
) -> AccountPhoneapp:
|
||||
for field, value in obj_in.dict(exclude_unset=True).items():
|
||||
setattr(db_obj, field, value)
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def delete(self, db: Session, acid: int) -> None:
|
||||
obj = db.query(AccountPhoneapp).filter(AccountPhoneapp.acid == acid).first()
|
||||
if obj:
|
||||
db.delete(obj)
|
||||
db.commit()
|
||||
|
||||
class CRUDAccountToutiaoapp:
|
||||
def create(self, db: Session, obj_in: AccountToutiaoappCreate) -> AccountToutiaoapp:
|
||||
db_obj = AccountToutiaoapp(**obj_in.dict())
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def get(self, db: Session, acid: int) -> Optional[AccountToutiaoapp]:
|
||||
return db.query(AccountToutiaoapp).filter(AccountToutiaoapp.acid == acid).first()
|
||||
|
||||
def update(
|
||||
self, db: Session, *, db_obj: AccountToutiaoapp, obj_in: AccountToutiaoappUpdate
|
||||
) -> AccountToutiaoapp:
|
||||
for field, value in obj_in.dict(exclude_unset=True).items():
|
||||
setattr(db_obj, field, value)
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def delete(self, db: Session, acid: int) -> None:
|
||||
obj = db.query(AccountToutiaoapp).filter(AccountToutiaoapp.acid == acid).first()
|
||||
if obj:
|
||||
db.delete(obj)
|
||||
db.commit()
|
||||
|
||||
|
||||
class CRUDAccount:
|
||||
class CRUDAccount(CRUDBase[Account, None, None]):
|
||||
def get_by_uniacid(self, db: Session, *, uniacid: int) -> Optional[Account]:
|
||||
return db.query(Account).filter(Account.uniacid == uniacid).first()
|
||||
return db.query(self.model).filter(self.model.uniacid == uniacid).first()
|
||||
|
||||
account = CRUDAccount()
|
||||
|
||||
|
||||
|
||||
|
||||
class CRUDAccountWebapp:
|
||||
def create(self, db: Session, *, obj_in: AccountWebappCreate) -> AccountWebapp:
|
||||
db_obj = AccountWebapp(**obj_in.dict())
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def get(self, db: Session, acid: int) -> Optional[AccountWebapp]:
|
||||
return db.query(AccountWebapp).filter(AccountWebapp.acid == acid).first()
|
||||
|
||||
def update(
|
||||
self,
|
||||
db: Session,
|
||||
*,
|
||||
db_obj: AccountWebapp,
|
||||
obj_in: AccountWebappUpdate
|
||||
) -> AccountWebapp:
|
||||
data = obj_in.dict(exclude_unset=True)
|
||||
for field, value in data.items():
|
||||
setattr(db_obj, field, value)
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def delete(self, db: Session, acid: int) -> None:
|
||||
obj = db.query(AccountWebapp).filter(AccountWebapp.acid == acid).first()
|
||||
if obj:
|
||||
db.delete(obj)
|
||||
db.commit()
|
||||
|
||||
|
||||
|
||||
class CRUDAccountWxapp:
|
||||
def create(self, db: Session, obj_in: AccountWxappCreate) -> AccountWxapp:
|
||||
db_obj = AccountWxapp(**obj_in.dict())
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def get(self, db: Session, acid: int) -> Optional[AccountWxapp]:
|
||||
return db.query(AccountWxapp).filter(AccountWxapp.acid == acid).first()
|
||||
|
||||
def update(
|
||||
self, db: Session, *, db_obj: AccountWxapp, obj_in: AccountWxappUpdate
|
||||
) -> AccountWxapp:
|
||||
for field, value in obj_in.dict(exclude_unset=True).items():
|
||||
setattr(db_obj, field, value)
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def delete(self, db: Session, acid: int) -> None:
|
||||
obj = db.query(AccountWxapp).filter(AccountWxapp.acid == acid).first()
|
||||
if obj:
|
||||
db.delete(obj)
|
||||
db.commit()
|
||||
|
||||
class CRUDAccountXzapp:
|
||||
def create(self, db: Session, obj_in: AccountXzappCreate) -> AccountXzapp:
|
||||
db_obj = AccountXzapp(**obj_in.dict())
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def get(self, db: Session, acid: int) -> Optional[AccountXzapp]:
|
||||
return db.query(AccountXzapp).filter(AccountXzapp.acid == acid).first()
|
||||
|
||||
def update(
|
||||
self, db: Session, *, db_obj: AccountXzapp, obj_in: AccountXzappUpdate
|
||||
) -> AccountXzapp:
|
||||
for field, value in obj_in.dict(exclude_unset=True).items():
|
||||
setattr(db_obj, field, value)
|
||||
db.add(db_obj)
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
def delete(self, db: Session, acid: int) -> None:
|
||||
obj = db.query(AccountXzapp).filter(AccountXzapp.acid == acid).first()
|
||||
if obj:
|
||||
db.delete(obj)
|
||||
db.commit()
|
||||
|
||||
# 在文件末尾添加实例
|
||||
account_wxapp = CRUDAccountWxapp()
|
||||
account_xzapp = CRUDAccountXzapp()
|
||||
account_webapp = CRUDAccountWebapp()
|
||||
|
||||
account_wechats = CRUDAccountWechats()
|
||||
ims_account_aliapp = CRUDAccountAliapp()
|
||||
ims_account_baiduapp = CRUDAccountBaiduapp()
|
||||
ims_account_phoneapp = CRUDAccountPhoneapp()
|
||||
ims_account_toutiaoapp = CRUDAccountToutiaoapp()
|
||||
# 实例化所有CRUD对象
|
||||
account = CRUDAccount(Account)
|
||||
account_wechats = CRUDAccountWechats(AccountWechats)
|
||||
account_aliapp = CRUDAccountAliapp(AccountAliapp)
|
||||
account_baiduapp = CRUDAccountBaiduapp(AccountBaiduapp)
|
||||
account_phoneapp = CRUDAccountPhoneapp(AccountPhoneapp)
|
||||
account_toutiaoapp = CRUDAccountToutiaoapp(AccountToutiaoapp)
|
||||
account_webapp = CRUDAccountWebapp(AccountWebapp)
|
||||
account_wxapp = CRUDAccountWxapp(AccountWxapp)
|
||||
account_xzapp = CRUDAccountXzapp(AccountXzapp)
|
||||
73
mooc/crud/crud_base.py
Normal file
73
mooc/crud/crud_base.py
Normal file
@ -0,0 +1,73 @@
|
||||
from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union
|
||||
from fastapi.encoders import jsonable_encoder
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy.orm import Session
|
||||
from mooc.db.database import Base
|
||||
|
||||
ModelType = TypeVar("ModelType", bound=Base)
|
||||
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
|
||||
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
|
||||
|
||||
class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
|
||||
def __init__(self, model: Type[ModelType]):
|
||||
"""
|
||||
CRUD base class
|
||||
Args:
|
||||
model: SQLAlchemy模型类
|
||||
"""
|
||||
self.model = model
|
||||
|
||||
def get(self, db: Session, id: Any) -> Optional[ModelType]:
|
||||
return db.query(self.model).filter(self.model.acid == id).first()
|
||||
|
||||
def get_multi(
|
||||
self, db: Session, *, skip: int = 0, limit: int = 100
|
||||
) -> List[ModelType]:
|
||||
return db.query(self.model).offset(skip).limit(limit).all()
|
||||
|
||||
def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType:
|
||||
obj_in_data = jsonable_encoder(obj_in)
|
||||
db_obj = self.model(**obj_in_data)
|
||||
db.add(db_obj)
|
||||
try:
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
raise e
|
||||
return db_obj
|
||||
|
||||
def update(
|
||||
self,
|
||||
db: Session,
|
||||
*,
|
||||
db_obj: ModelType,
|
||||
obj_in: Union[UpdateSchemaType, Dict[str, Any]]
|
||||
) -> ModelType:
|
||||
obj_data = jsonable_encoder(db_obj)
|
||||
if isinstance(obj_in, dict):
|
||||
update_data = obj_in
|
||||
else:
|
||||
update_data = obj_in.dict(exclude_unset=True)
|
||||
for field in obj_data:
|
||||
if field in update_data:
|
||||
setattr(db_obj, field, update_data[field])
|
||||
db.add(db_obj)
|
||||
try:
|
||||
db.commit()
|
||||
db.refresh(db_obj)
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
raise e
|
||||
return db_obj
|
||||
|
||||
def delete(self, db: Session, *, id: int) -> ModelType:
|
||||
obj = db.query(self.model).get(id)
|
||||
if obj:
|
||||
try:
|
||||
db.delete(obj)
|
||||
db.commit()
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
raise e
|
||||
return obj
|
||||
Loading…
Reference in New Issue
Block a user