-
Notifications
You must be signed in to change notification settings - Fork 18
Open
Description
@with_session
async def add(
self, table_obj: [T_BaseOrmTable, dict], *, orm_table: Type[BaseOrmTable] = None, session: AsyncSession = None
) -> int:
"""
插入一条数据
Args:
table_obj: orm映射类实例对象, eg. UserTable(username="hui", age=18) or {"username": "hui", "age": 18}
orm_table: orm表映射类
session: 数据库会话对象,如果为 None,则通过装饰器在方法内部开启新的事务
Returns: 新增的id
table_obj.id
"""
session = session or self.session
orm_table = orm_table or self.orm_table
if isinstance(table_obj, dict):
table_obj = orm_table(**table_obj)
session.add(table_obj)
await session.flush(objects=[table_obj]) # 刷新对象状态,获取新增的id
return table_obj.id
下面是我修改的一版,请大佬斧正:
@with_session
async def add(
self,
table_obj: [BaseOrmTable, dict],
*,
orm_table: Type[BaseOrmTable] = None,
session: AsyncSession = None
):
"""
单条插入数据库中
:param table_obj:
:param orm_table:
:param session:
:return:
"""
session = session or self.session
orm_table = orm_table or self.orm_table
if isinstance(table_obj, dict):
table_obj = orm_table(**table_obj)
session.add(table_obj)
await session.flush(objects=[table_obj])
primary_key_name = getattr(inspect(orm_table).primary_key[0], 'name')
return getattr(table_obj, primary_key_name)
Reactions are currently unavailable
Metadata
Metadata
Assignees
Labels
No labels