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)