一、说明
上图是做了个登录接口，很简单很简单，只是为了跑通代码
证明接口文档正常，
数据库正常
我们贴下代码
安装dao==》server==》controler
二、建立部分do文件
需要学习do vo这些区别的
可以看看
领域驱动系列-浅析VO、DTO、DO、PO - 知乎 (zhihu.com)
https://zhuanlan.zhihu.com/p/643346204
#mucauna-fastapi/admin/entity/do/dept_do.py
from sqlalchemy import Column, Integer, String, DateTime
from config.database import Base
from datetime import datetime
class SysDept(Base):
"""
部门表
"""
__tablename__ = 'sys_dept'dept_id = Column(Integer, primary_key=True, autoincrement=True, comment='部门id')
parent_id = Column(Integer, default=0, comment='父部门id')
ancestors = Column(String(50), nullable=True, default='', comment='祖级列表')
dept_name = Column(String(30), nullable=True, default='', comment='部门名称')
order_num = Column(Integer, default=0, comment='显示顺序')
leader = Column(String(20), nullable=True, default=None, comment='负责人')
phone = Column(String(11), nullable=True, default=None, comment='联系电话')
email = Column(String(50), nullable=True, default=None, comment='邮箱')
status = Column(String(1), nullable=True, default=0, comment='部门状态（0正常 1停用）')
del_flag = Column(String(1), nullable=True, default=0, comment='删除标志（0代表存在 2代表删除）')
create_by = Column(String(64), nullable=True, default='', comment='创建者')
create_time = Column(DateTime, nullable=True, default=datetime.now(), comment='创建时间')
update_by = Column(String(64), nullable=True, default='', comment='更新者')
update_time = Column(DateTime, nullable=True, default=datetime.now(), comment='更新时间')
#mucauna-fastapi/admin/entity/do/user_do.py
from sqlalchemy import Column, Integer, String, DateTime
from config.database import Base
from datetime import datetime
class SysUser(Base):
"""
用户信息表
"""
__tablename__ = 'sys_user'user_id = Column(Integer, primary_key=True, autoincrement=True, comment='用户ID')
dept_id = Column(Integer, comment='部门ID')
user_name = Column(String(30, collation='utf8_general_ci'), nullable=False, comment='用户账号')
nick_name = Column(String(30, collation='utf8_general_ci'), nullable=False, comment='用户昵称')
user_type = Column(String(2, collation='utf8_general_ci'), default='00', comment='用户类型（00系统用户）')
email = Column(String(50, collation='utf8_general_ci'), default='', comment='用户邮箱')
phonenumber = Column(String(11, collation='utf8_general_ci'), default='', comment='手机号码')
sex = Column(String(1, collation='utf8_general_ci'), default='0', comment='用户性别（0男 1女 2未知）')
avatar = Column(String(100, collation='utf8_general_ci'), default='', comment='头像地址')
password = Column(String(100, collation='utf8_general_ci'), default='', comment='密码')
status = Column(String(1, collation='utf8_general_ci'), default='0', comment='帐号状态（0正常 1停用）')
del_flag = Column(String(1, collation='utf8_general_ci'), default='0', comment='删除标志（0代表存在 2代表删除）')
login_ip = Column(String(128, collation='utf8_general_ci'), default='', comment='最后登录IP')
login_date = Column(DateTime, comment='最后登录时间')
create_by = Column(String(64, collation='utf8_general_ci'), default='', comment='创建者')
create_time = Column(DateTime, comment='创建时间', default=datetime.now())
update_by = Column(String(64, collation='utf8_general_ci'), default='', comment='更新者')
update_time = Column(DateTime, comment='更新时间', default=datetime.now())
remark = Column(String(500, collation='utf8_general_ci'), comment='备注')
class SysUserRole(Base):
"""
用户和角色关联表
"""
__tablename__ = 'sys_user_role'user_id = Column(Integer, primary_key=True, nullable=False, comment='用户ID')
role_id = Column(Integer, primary_key=True, nullable=False, comment='角色ID')
class SysUserPost(Base):
"""
用户与岗位关联表
"""
__tablename__ = 'sys_user_post'user_id = Column(Integer, primary_key=True, nullable=False, comment='用户ID')
post_id = Column(Integer, primary_key=True, nullable=False, comment='岗位ID')
三、建立部分vo文件
#mucauna-fastapi/admin/entity/vo/login_vo.py
from pydantic import BaseModel, ConfigDict
from pydantic.alias_generators import to_camel
from typing import Optional
class UserLogin(BaseModel):
model_config = ConfigDict(alias_generator=to_camel)user_name: str
password: str
code: Optional[str] = None
uuid: Optional[str] = None
login_info: Optional[dict] = None
captcha_enabled: Optional[bool] = None
四、login_dao
#mucauna-fastapi/admin/dao/login_dao.py
from sqlalchemy.orm import Session
from admin.entity.do.user_do import SysUser
from admin.entity.do.dept_do import SysDept
from sqlalchemy import and_def login_by_account(db: Session, user_name: str):
"""
根据用户名查询用户信息
:param db: orm对象
:param user_name: 用户名
:return: 用户对象
"""
user = db.query(SysUser, SysDept)\
.filter(SysUser.user_name == user_name, SysUser.del_flag == '0') \
.outerjoin(SysDept, and_(SysUser.dept_id == SysDept.dept_id, SysDept.status == 0, SysDept.del_flag == 0)) \
.distinct() \
.first()return user
五、login_service
#mucauna-fastapi/admin/service/login_service.py
from fastapi import Request
from admin.entity.vo.login_vo import *
from admin.dao.login_dao import *
from systemhandle.exceptions.exception import LoginException
from loguru import loggerclass LoginService:
"""
登录模块服务层
"""
@classmethod
async def authenticate_user(cls, request: Request, query_db: Session, login_user: UserLogin):
"""
根据用户名密码校验用户登录
:param request: Request对象
:param query_db: orm对象
:param login_user: 登录用户对象
:return: 校验结果
"""
user = login_by_account(query_db, login_user.user_name)
if not user:
logger.warning("用户不存在")
raise LoginException(data="", message="用户不存在")
logger.debug("user = ",user)
六、login_controller
#mucauna-fastapi/admin/controller/login_controller.py
from fastapi import APIRouter
from admin.service.login_service import *
from admin.entity.vo.login_vo import *
from admin.dao.login_dao import *from utils.response_util import ResponseUtil
from utils.log_util import *
from datetime import timedelta
from fastapi import Depends
from config.get_db import get_dbloginController = APIRouter()
@loginController.post("/login")
async def login(request: Request, userName:str,password:str, query_db: Session = Depends(get_db)):
user = UserLogin(
userName=userName,
password=password, #"$2a$10$7JB720yubVSZvUI0rEqK/.VqGOZTH.ulu33dHOiBE8ByOhJIrdAu2",
captchaEnabled=False
)
try:
result = await LoginService.authenticate_user(request, query_db, user)
logger.debug("result = ",result)
except LoginException as e:
return ResponseUtil.failure(msg=e.message)
return ResponseUtil.success(
msg='登录成功',
dict_content={'token': "1234"})
七、最后一步
在server.py上加上
from admin.controller.login_controller import loginController
。。。。。
# 加载路由列表
controller_list = [
#先全部去掉，后面再一个一个加
{'router': loginController, 'tags': ['登录模块']},
]
这几步基本算是加一个业务的逻辑，就这么加，后面再搞搞权限啊，什么的
我的目标是ai，这个只是打个基础，欢迎大家一起学习
