提交part2的crud操作 crud_core crud_users crud_userapi

Signed-off-by: 雨过 <zxx1747362695@qq.com>
This commit is contained in:
雨过 2025-01-05 16:29:20 +00:00 committed by Gitee
parent 81d9a0d5fd
commit 0106c68699
No known key found for this signature in database
GPG Key ID: 173E9B9CA92EEF8F
3 changed files with 471 additions and 0 deletions

181
mooc/crud/crud_core.py Normal file
View File

@ -0,0 +1,181 @@
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)

29
mooc/crud/crud_userapi.py Normal file
View File

@ -0,0 +1,29 @@
from typing import Optional
from sqlalchemy.orm import Session
from mooc.crud.crud_base import CRUDBase
from mooc.models.userapi import (
UserapiCache,
UserapiReply
)
from mooc.schemas.userapi import (
UserapiCacheCreate, UserapiCacheUpdate,
UserapiReplyCreate, UserapiReplyUpdate
)
class CRUDUserapiCache(CRUDBase[UserapiCache, UserapiCacheCreate, UserapiCacheUpdate]):
def get_by_key(self, db: Session, *, key: str) -> Optional[UserapiCache]:
return self.get_by_field(db, "key", key)
def get_by_id(self, db: Session, *, id: int) -> Optional[UserapiCache]:
return self.get_by_field(db, "id", id)
class CRUDUserapiReply(CRUDBase[UserapiReply, UserapiReplyCreate, UserapiReplyUpdate]):
def get_by_rid(self, db: Session, *, rid: int) -> Optional[UserapiReply]:
return self.get_by_field(db, "rid", rid)
def get_by_id(self, db: Session, *, id: int) -> Optional[UserapiReply]:
return self.get_by_field(db, "id", id)
# 创建实例
userapi_cache_crud = CRUDUserapiCache(UserapiCache)
userapi_reply_crud = CRUDUserapiReply(UserapiReply)

261
mooc/crud/crud_users.py Normal file
View File

@ -0,0 +1,261 @@
from typing import Optional
from sqlalchemy.orm import Session
from mooc.crud.crud_base import CRUDBase
from mooc.models.users import (
Users,
UsersBind,
UsersCreateGroup,
UsersGroup,
UsersInvitation,
UsersLastUse,
UsersLoginLogs,
UsersOperateHistory,
UsersOperateStar,
UsersPermission,
UsersProfile,
UsersExtraGroup,
UsersExtraLimit,
UsersExtraModules,
UsersExtraTemplates,
UsersFailedLogin,
UsersFounderGroup,
UsersFounderOwnCreateGroups,
UsersFounderOwnUniGroups,
UsersFounderOwnUsers,
UsersFounderOwnUsersGroups
)
from mooc.schemas.users import (
UsersCreate, UsersUpdate,
UsersBindCreate, UsersBindUpdate,
UsersCreateGroupCreate, UsersCreateGroupUpdate,
UsersGroupCreate, UsersGroupUpdate,
UsersInvitationCreate, UsersInvitationUpdate,
UsersLastUseCreate, UsersLastUseUpdate,
UsersLoginLogsCreate, UsersLoginLogsUpdate,
UsersOperateHistoryCreate, UsersOperateHistoryUpdate,
UsersOperateStarCreate, UsersOperateStarUpdate,
UsersPermissionCreate, UsersPermissionUpdate,
UsersProfileCreate, UsersProfileUpdate,
UsersExtraGroupCreate, UsersExtraGroupUpdate,
UsersExtraLimitCreate, UsersExtraLimitUpdate,
UsersExtraModulesCreate, UsersExtraModulesUpdate,
UsersExtraTemplatesCreate, UsersExtraTemplatesUpdate,
UsersFailedLoginCreate, UsersFailedLoginUpdate,
UsersFounderGroupCreate, UsersFounderGroupUpdate,
UsersFounderOwnCreateGroupsCreate, UsersFounderOwnCreateGroupsUpdate,
UsersFounderOwnUniGroupsCreate, UsersFounderOwnUniGroupsUpdate,
UsersFounderOwnUsersCreate, UsersFounderOwnUsersUpdate,
UsersFounderOwnUsersGroupsCreate, UsersFounderOwnUsersGroupsUpdate
)
class CRUDUsers(CRUDBase[Users, UsersCreate, UsersUpdate]):
def get_by_uid(self, db: Session, *, uid: int) -> Optional[Users]:
return self.get_by_field(db, "uid", uid)
def get_by_username(self, db: Session, *, username: str) -> Optional[Users]:
return self.get_by_field(db, "username", username)
class CRUDUsersBind(CRUDBase[UsersBind, UsersBindCreate, UsersBindUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersBind]:
return self.get_by_field(db, "id", id)
def get_by_uid(self, db: Session, *, uid: int) -> Optional[UsersBind]:
return self.get_by_field(db, "uid", uid)
class CRUDUsersCreateGroup(CRUDBase[UsersCreateGroup, UsersCreateGroupCreate, UsersCreateGroupUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersCreateGroup]:
return self.get_by_field(db, "id", id)
def get_by_group_name(self, db: Session, *, group_name: str) -> Optional[UsersCreateGroup]:
return self.get_by_field(db, "group_name", group_name)
class CRUDUsersExtraGroup(CRUDBase[UsersExtraGroup, UsersExtraGroupCreate, UsersExtraGroupUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersExtraGroup]:
return self.get_by_field(db, "id", id)
def get_by_uid(self, db: Session, *, uid: int) -> Optional[UsersExtraGroup]:
return self.get_by_field(db, "uid", uid)
class CRUDUsersExtraLimit(CRUDBase[UsersExtraLimit, UsersExtraLimitCreate, UsersExtraLimitUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersExtraLimit]:
return self.get_by_field(db, "id", id)
def get_by_uid(self, db: Session, *, uid: int) -> Optional[UsersExtraLimit]:
return self.get_by_field(db, "uid", uid)
class CRUDUsersExtraModules(CRUDBase[UsersExtraModules, UsersExtraModulesCreate, UsersExtraModulesUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersExtraModules]:
return self.get_by_field(db, "id", id)
def get_by_uid(self, db: Session, *, uid: int) -> Optional[UsersExtraModules]:
return self.get_by_field(db, "uid", uid)
class CRUDUsersExtraTemplates(CRUDBase[UsersExtraTemplates, UsersExtraTemplatesCreate, UsersExtraTemplatesUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersExtraTemplates]:
return self.get_by_field(db, "id", id)
def get_by_uid(self, db: Session, *, uid: int) -> Optional[UsersExtraTemplates]:
return self.get_by_field(db, "uid", uid)
class CRUDUsersFailedLogin(CRUDBase[UsersFailedLogin, UsersFailedLoginCreate, UsersFailedLoginUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersFailedLogin]:
return self.get_by_field(db, "id", id)
def get_by_username(self, db: Session, *, username: str) -> Optional[UsersFailedLogin]:
return self.get_by_field(db, "username", username)
class CRUDUsersFounderGroup(CRUDBase[UsersFounderGroup, UsersFounderGroupCreate, UsersFounderGroupUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersFounderGroup]:
return self.get_by_field(db, "id", id)
def get_by_name(self, db: Session, *, name: str) -> Optional[UsersFounderGroup]:
return self.get_by_field(db, "name", name)
class CRUDUsersFounderOwnCreateGroups(CRUDBase[UsersFounderOwnCreateGroups, UsersFounderOwnCreateGroupsCreate, UsersFounderOwnCreateGroupsUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersFounderOwnCreateGroups]:
return self.get_by_field(db, "id", id)
def get_by_founder_uid(self, db: Session, *, founder_uid: int) -> Optional[UsersFounderOwnCreateGroups]:
return self.get_by_field(db, "founder_uid", founder_uid)
### CRUD操作类 for UsersFounderOwnUniGroups表
class CRUDUsersFounderOwnUniGroups(CRUDBase[UsersFounderOwnUniGroups, UsersFounderOwnUniGroupsCreate, UsersFounderOwnUniGroupsUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersFounderOwnUniGroups]:
return self.get_by_field(db, "id", id)
def get_by_founder_uid(self, db: Session, *, founder_uid: int) -> Optional[UsersFounderOwnUniGroups]:
return self.get_by_field(db, "founder_uid", founder_uid)
### CRUD操作类 for UsersFounderOwnUsers表
class CRUDUsersFounderOwnUsers(CRUDBase[UsersFounderOwnUsers, UsersFounderOwnUsersCreate, UsersFounderOwnUsersUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersFounderOwnUsers]:
return self.get_by_field(db, "id", id)
def get_by_uid(self, db: Session, *, uid: int) -> Optional[UsersFounderOwnUsers]:
return self.get_by_field(db, "uid", uid)
### CRUD操作类 for UsersFounderOwnUsersGroups表
class CRUDUsersFounderOwnUsersGroups(CRUDBase[UsersFounderOwnUsersGroups, UsersFounderOwnUsersGroupsCreate, UsersFounderOwnUsersGroupsUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersFounderOwnUsersGroups]:
return self.get_by_field(db, "id", id)
def get_by_founder_uid(self, db: Session, *, founder_uid: int) -> Optional[UsersFounderOwnUsersGroups]:
return self.get_by_field(db, "founder_uid", founder_uid)
### CRUD操作类 for UsersGroup表
class CRUDUsersGroup(CRUDBase[UsersGroup, UsersGroupCreate, UsersGroupUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersGroup]:
return self.get_by_field(db, "id", id)
def get_by_name(self, db: Session, *, name: str) -> Optional[UsersGroup]:
return self.get_by_field(db, "name", name)
### CRUD操作类 for UsersInvitation表
class CRUDUsersInvitation(CRUDBase[UsersInvitation, UsersInvitationCreate, UsersInvitationUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersInvitation]:
return self.get_by_field(db, "id", id)
def get_by_code(self, db: Session, *, code: str) -> Optional[UsersInvitation]:
return self.get_by_field(db, "code", code)
### CRUD操作类 for UsersLastuse表
class CRUDUsersLastuse(CRUDBase[UsersLastUse, UsersLastUseCreate, UsersLastUseUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersLastUse]:
return self.get_by_field(db, "id", id)
def get_by_uid(self, db: Session, *, uid: int) -> Optional[UsersLastUse]:
return self.get_by_field(db, "uid", uid)
### CRUD操作类 for UsersLoginLogs表
class CRUDUsersLoginLogs(CRUDBase[UsersLoginLogs, UsersLoginLogsCreate, UsersLoginLogsUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersLoginLogs]:
return self.get_by_field(db, "id", id)
def get_by_uid(self, db: Session, *, uid: int) -> Optional[UsersLoginLogs]:
return self.get_by_field(db, "uid", uid)
### CRUD操作类 for UsersOperateHistory表
class CRUDUsersOperateHistory(CRUDBase[UsersOperateHistory, UsersOperateHistoryCreate, UsersOperateHistoryUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersOperateHistory]:
return self.get_by_field(db, "id", id)
def get_by_uid(self, db: Session, *, uid: int) -> Optional[UsersOperateHistory]:
return self.get_by_field(db, "uid", uid)
### CRUD操作类 for UsersOperateStar表
class CRUDUsersOperateStar(CRUDBase[UsersOperateStar, UsersOperateStarCreate, UsersOperateStarUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersOperateStar]:
return self.get_by_field(db, "id", id)
def get_by_uid(self, db: Session, *, uid: int) -> Optional[UsersOperateStar]:
return self.get_by_field(db, "uid", uid)
### CRUD操作类 for UsersPermission表
class CRUDUsersPermission(CRUDBase[UsersPermission, UsersPermissionCreate, UsersPermissionUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersPermission]:
return self.get_by_field(db, "id", id)
def get_by_uid(self, db: Session, *, uid: int) -> Optional[UsersPermission]:
return self.get_by_field(db, "uid", uid)
### CRUD操作类 for UsersProfile表
class CRUDUsersProfile(CRUDBase[UsersProfile, UsersProfileCreate, UsersProfileUpdate]):
def get_by_id(self, db: Session, *, id: int) -> Optional[UsersProfile]:
return self.get_by_field(db, "id", id)
def get_by_uid(self, db: Session, *, uid: int) -> Optional[UsersProfile]:
return self.get_by_field(db, "uid", uid)
# 创建实例
users_operate_star_crud = CRUDUsersOperateStar(UsersOperateStar)
users_permission_crud = CRUDUsersPermission(UsersPermission)
users_profile_crud = CRUDUsersProfile(UsersProfile)
# 创建实例
users_lastuse_crud = CRUDUsersLastuse(UsersLastUse)
users_login_logs_crud = CRUDUsersLoginLogs(UsersLoginLogs)
users_operate_history_crud = CRUDUsersOperateHistory(UsersOperateHistory)
# 创建实例
users_founder_own_uni_groups_crud = CRUDUsersFounderOwnUniGroups(UsersFounderOwnUniGroups)
users_founder_own_users_crud = CRUDUsersFounderOwnUsers(UsersFounderOwnUsers)
users_founder_own_users_groups_crud = CRUDUsersFounderOwnUsersGroups(UsersFounderOwnUsersGroups)
users_group_crud = CRUDUsersGroup(UsersGroup)
users_invitation_crud = CRUDUsersInvitation(UsersInvitation)
# 创建实例
users_extra_limit_crud = CRUDUsersExtraLimit(UsersExtraLimit)
users_extra_modules_crud = CRUDUsersExtraModules(UsersExtraModules)
users_extra_templates_crud = CRUDUsersExtraTemplates(UsersExtraTemplates)
users_failed_login_crud = CRUDUsersFailedLogin(UsersFailedLogin)
users_founder_group_crud = CRUDUsersFounderGroup(UsersFounderGroup)
users_founder_own_create_groups_crud = CRUDUsersFounderOwnCreateGroups(UsersFounderOwnCreateGroups)
# 创建实例
users_crud = CRUDUsers(Users)
users_bind_crud = CRUDUsersBind(UsersBind)
users_create_group_crud = CRUDUsersCreateGroup(UsersCreateGroup)
users_extra_group_crud = CRUDUsersExtraGroup(UsersExtraGroup)