当前位置:   article > 正文

FastAPI(五)sqlalchemy操作mysql数据库_fastapi mysql

fastapi mysql

一、安装依赖

pip install SQLAlchemy==1.4.46
pip install sqlacodegen
pip install pymysql

二、在根目录新建文件夹db

三、在db文件夹中新建session.py文件

  1. import os
  2. from contextlib import contextmanager
  3. from typing import Generator
  4. from urllib import parse
  5. from sqlalchemy import create_engine
  6. from sqlalchemy.orm import sessionmaker
  7. HOST = 'localhost'
  8. PORT = 3306
  9. USERNAME = 'root'
  10. PASSWORD = parse.quote_plus('123456')
  11. DB = 'authcenter'
  12. DB_URI = f'mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DB}'
  13. engine = create_engine(DB_URI, pool_pre_ping=True)
  14. SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
  15. # 这里一定要用with上下文去管理session,否则会出现很多诡异的情况!!!切记
  16. # 创建session
  17. DbSession = sessionmaker(bind=engine)
  18. db_session = DbSession()
  19. @contextmanager
  20. def session_maker(session=db_session):
  21. try:
  22. yield session
  23. session.commit()
  24. except:
  25. session.rollback()
  26. raise
  27. finally:
  28. session.close()
  29. def get_db() -> Generator:
  30. """
  31. get SQLAlchemy session to curd
  32. :return: SQLAlchemy Session
  33. """
  34. db = None
  35. try:
  36. db = SessionLocal()
  37. yield db
  38. finally:
  39. if db:
  40. db.close()
  41. def get_db_connect() -> Generator:
  42. """
  43. get SQLAlchemy connect to exec sql
  44. :return: SQLAlchemy connect
  45. """
  46. conn = None
  47. try:
  48. conn = engine.connect()
  49. yield conn
  50. finally:
  51. if conn:
  52. conn.close()
  53. # 逆向工程 自动生成模型文件
  54. if __name__ == '__main__':
  55. tables = ["user"]
  56. os.system(
  57. f'sqlacodegen --tables {",".join(tables)} {DB_URI} > db_models.py')

 温馨提示:在填写PASSWORD的密码栏的时候使用parse.quote_plus('123456')将密码包裹起来。这是防止有些时候密码中包含了特殊字符@,这个和驱动中的@冲突。

四、 执行session.py生成模型文件

五、CRUD

在user文件夹下新建crud文件夹,并在该文件夹下新建user_crud.py文件内容如下:

  1. from sqlalchemy.orm import Session
  2. from db.db_models import User
  3. def getByUsername(db: Session, *, username: str):
  4. """
  5. 根据用户名查询用户
  6. :param db:
  7. :param username:
  8. :return:
  9. """
  10. return db.query(User).filter(User.username == username).first()

六、调用CRUD

修改user文件夹下的views.py如下:

  1. from fastapi import APIRouter, Depends
  2. from sqlalchemy.orm import Session
  3. from apps.user.crud import user_crud
  4. from apps.user.schemas.user_schemas import Item
  5. from db.session import get_db
  6. router = APIRouter()
  7. @router.post("/login", summary="用户登录", tags=['用户'])
  8. async def login(item: Item, db: Session = Depends(get_db)):
  9. username = user_crud.getByUsername(db=db, username=item.username)
  10. return {'message': '登录成功', 'data': username}

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/码创造者/article/detail/975260
推荐阅读
相关标签
  

闽ICP备14008679号