181 lines
7.4 KiB
Python
181 lines
7.4 KiB
Python
from typing import Optional
|
|
from sqlalchemy.orm import Session
|
|
from mooc.crud.crud_base import CRUDBase
|
|
from mooc.models.core import (
|
|
CoreAttachment,
|
|
CoreCache,
|
|
CoreCron,
|
|
CoreCronRecord,
|
|
CoreJob,
|
|
CoreMenu,
|
|
CoreMenuShortcut,
|
|
CorePaylog,
|
|
CorePerformance,
|
|
CoreQueue,
|
|
CoreRefundlog,
|
|
CoreResource,
|
|
CoreSendsmsLog,
|
|
CoreSessions,
|
|
CoreSettings
|
|
)
|
|
from mooc.schemas.core import (
|
|
CoreAttachmentCreate, CoreAttachmentUpdate,
|
|
CoreCacheCreate, CoreCacheUpdate,
|
|
CoreCronCreate, CoreCronUpdate,
|
|
CoreCronRecordCreate, CoreCronRecordUpdate,
|
|
CoreJobCreate, CoreJobUpdate,
|
|
CoreMenuCreate, CoreMenuUpdate,
|
|
CoreMenuShortcutCreate, CoreMenuShortcutUpdate,
|
|
CorePaylogCreate, CorePaylogUpdate,
|
|
CorePerformanceCreate, CorePerformanceUpdate,
|
|
CoreQueueCreate, CoreQueueUpdate,
|
|
CoreRefundlogCreate, CoreRefundlogUpdate,
|
|
CoreResourceCreate, CoreResourceUpdate,
|
|
CoreSendsmsLogCreate, CoreSendsmsLogUpdate,
|
|
CoreSessionsCreate, CoreSessionsUpdate,
|
|
CoreSettingsCreate, CoreSettingsUpdate
|
|
)
|
|
|
|
### CRUD操作类 for CoreAttachment表
|
|
class CRUDCoreAttachment(CRUDBase[CoreAttachment, CoreAttachmentCreate, CoreAttachmentUpdate]):
|
|
def get_by_id(self, db: Session, *, id: int) -> Optional[CoreAttachment]:
|
|
return self.get_by_field(db, "id", id)
|
|
|
|
|
|
### CRUD操作类 for CoreCache表
|
|
class CRUDCoreCache(CRUDBase[CoreCache, CoreCacheCreate, CoreCacheUpdate]):
|
|
def get_by_key(self, db: Session, *, key: str) -> Optional[CoreCache]:
|
|
return self.get_by_field(db, "key", key)
|
|
|
|
|
|
### CRUD操作类 for CoreCron表
|
|
class CRUDCoreCron(CRUDBase[CoreCron, CoreCronCreate, CoreCronUpdate]):
|
|
def get_by_id(self, db: Session, *, id: int) -> Optional[CoreCron]:
|
|
return self.get_by_field(db, "id", id)
|
|
|
|
def get_by_uniacid(self, db: Session, *, uniacid: int) -> Optional[CoreCron]:
|
|
return self.get_by_field(db, "uniacid", uniacid)
|
|
|
|
|
|
### CRUD操作类 for CoreCronRecord表
|
|
class CRUDCoreCronRecord(CRUDBase[CoreCronRecord, CoreCronRecordCreate, CoreCronRecordUpdate]):
|
|
def get_by_id(self, db: Session, *, id: int) -> Optional[CoreCronRecord]:
|
|
return self.get_by_field(db, "id", id)
|
|
|
|
def get_by_uniacid(self, db: Session, *, uniacid: int) -> Optional[CoreCronRecord]:
|
|
return self.get_by_field(db, "uniacid", uniacid)
|
|
|
|
|
|
### CRUD操作类 for CoreJob表
|
|
class CRUDCoreJob(CRUDBase[CoreJob, CoreJobCreate, CoreJobUpdate]):
|
|
def get_by_id(self, db: Session, *, id: int) -> Optional[CoreJob]:
|
|
return self.get_by_field(db, "id", id)
|
|
|
|
|
|
### CRUD操作类 for CoreMenu表
|
|
class CRUDCoreMenu(CRUDBase[CoreMenu, CoreMenuCreate, CoreMenuUpdate]):
|
|
def get_by_id(self, db: Session, *, id: int) -> Optional[CoreMenu]:
|
|
return self.get_by_field(db, "id", id)
|
|
|
|
|
|
### CRUD操作类 for CoreMenuShortcut表
|
|
class CRUDCoreMenuShortcut(CRUDBase[CoreMenuShortcut, CoreMenuShortcutCreate, CoreMenuShortcutUpdate]):
|
|
def get_by_id(self, db: Session, *, id: int) -> Optional[CoreMenuShortcut]:
|
|
return self.get_by_field(db, "id", id)
|
|
|
|
def get_by_uid(self, db: Session, *, uid: int) -> Optional[CoreMenuShortcut]:
|
|
return self.get_by_field(db, "uid", uid)
|
|
|
|
|
|
### CRUD操作类 for CorePaylog表
|
|
class CRUDCorePaylog(CRUDBase[CorePaylog, CorePaylogCreate, CorePaylogUpdate]):
|
|
def get_by_plid(self, db: Session, *, plid: int) -> Optional[CorePaylog]:
|
|
return self.get_by_field(db, "plid", plid)
|
|
|
|
def get_by_openid(self, db: Session, *, openid: str) -> Optional[CorePaylog]:
|
|
return self.get_by_field(db, "openid", openid)
|
|
|
|
def get_by_tid(self, db: Session, *, tid: int) -> Optional[CorePaylog]:
|
|
return self.get_by_field(db, "tid", tid)
|
|
|
|
def get_by_uniacid(self, db: Session, *, uniacid: int) -> Optional[CorePaylog]:
|
|
return self.get_by_field(db, "uniacid", uniacid)
|
|
|
|
|
|
### CRUD操作类 for CorePerformance表
|
|
class CRUDCorePerformance(CRUDBase[CorePerformance, CorePerformanceCreate, CorePerformanceUpdate]):
|
|
def get_by_id(self, db: Session, *, id: int) -> Optional[CorePerformance]:
|
|
return self.get_by_field(db, "id", id)
|
|
|
|
|
|
### CRUD操作类 for CoreQueue表
|
|
class CRUDCoreQueue(CRUDBase[CoreQueue, CoreQueueCreate, CoreQueueUpdate]):
|
|
def get_by_qid(self, db: Session, *, qid: int) -> Optional[CoreQueue]:
|
|
return self.get_by_field(db, "qid", qid)
|
|
|
|
def get_by_uniacid_and_acid(self, db: Session, *, uniacid: int, acid: int) -> Optional[CoreQueue]:
|
|
return self.get_by_field(db, "uniacid", uniacid, "acid", acid)
|
|
|
|
def get_by_module(self, db: Session, *, module: str) -> Optional[CoreQueue]:
|
|
return self.get_by_field(db, "module", module)
|
|
|
|
|
|
### CRUD操作类 for CoreRefundlog表
|
|
class CRUDCoreRefundlog(CRUDBase[CoreRefundlog, CoreRefundlogCreate, CoreRefundlogUpdate]):
|
|
def get_by_id(self, db: Session, *, id: int) -> Optional[CoreRefundlog]:
|
|
return self.get_by_field(db, "id", id)
|
|
|
|
def get_by_refund_uniontid(self, db: Session, *, refund_uniontid: str) -> Optional[CoreRefundlog]:
|
|
return self.get_by_field(db, "refund_uniontid", refund_uniontid)
|
|
|
|
def get_by_uniontid(self, db: Session, *, uniontid: str) -> Optional[CoreRefundlog]:
|
|
return self.get_by_field(db, "uniontid", uniontid)
|
|
|
|
|
|
### CRUD操作类 for CoreResource表
|
|
class CRUDCoreResource(CRUDBase[CoreResource, CoreResourceCreate, CoreResourceUpdate]):
|
|
def get_by_mid(self, db: Session, *, mid: int) -> Optional[CoreResource]:
|
|
return self.get_by_field(db, "mid", mid)
|
|
|
|
def get_by_uniacid(self, db: Session, *, uniacid: int) -> Optional[CoreResource]:
|
|
return self.get_by_field(db, "uniacid", uniacid)
|
|
|
|
def get_by_type(self, db: Session, *, type: str) -> Optional[CoreResource]:
|
|
return self.get_by_field(db, "type", type)
|
|
|
|
|
|
### CRUD操作类 for CoreSendsmsLog表
|
|
class CRUDCoreSendsmsLog(CRUDBase[CoreSendsmsLog, CoreSendsmsLogCreate, CoreSendsmsLogUpdate]):
|
|
def get_by_id(self, db: Session, *, id: int) -> Optional[CoreSendsmsLog]:
|
|
return self.get_by_field(db, "id", id)
|
|
|
|
|
|
### CRUD操作类 for CoreSessions表
|
|
class CRUDCoreSessions(CRUDBase[CoreSessions, CoreSessionsCreate, CoreSessionsUpdate]):
|
|
def get_by_sid(self, db: Session, *, sid: str) -> Optional[CoreSessions]:
|
|
return self.get_by_field(db, "sid", sid)
|
|
|
|
|
|
### CRUD操作类 for CoreSettings表
|
|
class CRUDCoreSettings(CRUDBase[CoreSettings, CoreSettingsCreate, CoreSettingsUpdate]):
|
|
def get_by_key(self, db: Session, *, key: str) -> Optional[CoreSettings]:
|
|
return self.get_by_field(db, "key", key)
|
|
|
|
|
|
# 创建实例
|
|
core_paylog_crud = CRUDCorePaylog(CorePaylog)
|
|
core_performance_crud = CRUDCorePerformance(CorePerformance)
|
|
core_queue_crud = CRUDCoreQueue(CoreQueue)
|
|
core_refundlog_crud = CRUDCoreRefundlog(CoreRefundlog)
|
|
core_resource_crud = CRUDCoreResource(CoreResource)
|
|
core_sendsms_log_crud = CRUDCoreSendsmsLog(CoreSendsmsLog)
|
|
core_sessions_crud = CRUDCoreSessions(CoreSessions)
|
|
core_settings_crud = CRUDCoreSettings(CoreSettings)
|
|
# 创建实例
|
|
core_attachment_crud = CRUDCoreAttachment(CoreAttachment)
|
|
core_cache_crud = CRUDCoreCache(CoreCache)
|
|
core_cron_crud = CRUDCoreCron(CoreCron)
|
|
core_cron_record_crud = CRUDCoreCronRecord(CoreCronRecord)
|
|
core_job_crud = CRUDCoreJob(CoreJob)
|
|
core_menu_crud = CRUDCoreMenu(CoreMenu)
|
|
core_menu_shortcut_crud = CRUDCoreMenuShortcut(CoreMenuShortcut) |