You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
from sqlalchemy import Column, BigInteger, Select, Delete, Update
|
|
|
|
from common.global_enums import IsDelete
|
|
from utils import current_timestamp
|
|
|
|
# 上下文变量，控制是否启用逻辑删除过滤
|
|
_SOFT_DELETE_ENABLED = True
|
|
from sqlmodel import SQLModel, Field
|
|
|
|
|
|
class DbBaseModel(SQLModel, table=False):
|
|
id: str = Field(default=None, max_length=32, primary_key=True)
|
|
created_time: int = Field(
|
|
default_factory=current_timestamp,
|
|
sa_type=BigInteger,
|
|
description="创建时间"
|
|
)
|
|
# created_by = CharField(max_length=32, index=True)
|
|
updated_time: int = Field(
|
|
default_factory=current_timestamp,
|
|
sa_type=BigInteger,
|
|
sa_column_kwargs={"onupdate": current_timestamp},
|
|
description="更新时间"
|
|
)
|
|
# updated_by = CharField(max_length=32)
|
|
is_deleted: int = Field(default=IsDelete.NO_DELETE)
|
|
|
|
# class Config:
|
|
# arbitrary_types_allowed = True
|
|
@classmethod
|
|
def select(cls, fields=None):
|
|
if fields is None:
|
|
fields = cls
|
|
return Select(fields)
|
|
|
|
|
|
@classmethod
|
|
def delete(cls):
|
|
return Delete(cls)
|
|
|
|
@classmethod
|
|
def update(cls):
|
|
return Update(cls)
|
|
|
|
|
|
@classmethod
|
|
def update_by_ids(cls, ids: list[str],update_dict: dict):
|
|
update_dict.pop("id",None)
|
|
return Update(cls).where(cls.id.in_(ids)).values(**update_dict)
|