Java博客作业
2026/10/13 7:02:11
python在开发Web项目常用框架
pip install fastapi pip install uvicorn pip install sqlalchemy pip install pymysqlfrom sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker #设置数据库的url SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:wangcheng@192.168.1.229:3306/test" #创建数据库引擎 engine = create_engine( SQLALCHEMY_DATABASE_URL, echo=True, #打印sql语句 # connect_args={"check_same_thread":False} #解决多线程问题 sqlite使用 ) #创建回话 sessionLocal = sessionmaker(autocommit=False,autoflush=False,bind=engine,expire_on_commit=True) #映射 Base = declarative_base()from sqlalchemy import Column,Integer,String from database import Base #定义User类,继承自Base class User(Base): #定义表名 __tablename__ = "user" #定义id字段,主键,自增,索引,可以为空 id = Column(Integer,primary_key=True,index=True,nullable=True,autoincrement=True) #定义name字段,不能为空,长度100 name = Column(String(100),nullable=False,comment="姓名") #定义gender字段,长度为10 gender = Column(String(10),comment="性别")from pydantic import BaseModel from typing import Union,Optional #定义userbase类,继承自basemodal class UserBase(BaseModel): #定义id属性,类型为int,默认值为1 id : int = 1 #定义Config属性,用于配置 class Config: #设置orm_mode属性,值为True orm_mode = True #创建用户类 class create_user(UserBase): #用户名 name: str #性别 gender: Optional[str] = None #定义一个config类 class Config: #定义一个orm_mode属性为true orm_mode = True class read_user(UserBase): #用户名 name: str gender: Optional[str] = None class Config: orm_mode = Truefrom sqlalchemy.orm import Session import model,schemas #查看用户信息 def get_user(db:Session,skip:int=0,limit:int=10): #返回skip和limit之间的用户信息 return db.query(model.User).offset(skip).limit(limit).all() #创建新用户 def create_user(db: Session,user:schemas.create_user): #创建一个User模型,并将加密后的密码赋值给password db_user = model.User(id = user.id, name = user.name, gender = user.gender) #将User模型添加到数据库 db.add(db_user) #提交数据库操作 db.commit() #刷新数据库 db.refresh(db_user) #返回创建的用户 return db_user def get_user_by_account(db: Session,name): return db.query(model.User).filter(model.User.name == name).all()from fastapi import Depends,APIRouter,HTTPException from sqlalchemy.orm import Session import crud,model,schemas from database import sessionLocal,engine #创建数据库表 model.Base.metadata.create_all(bind=engine) #创建所有模型 app_ms = APIRouter() #定义一个函数get_db,用于获取数据库连接 def get_db(): db = sessionLocal() try: yield db finally: db.close() #获取user数据 @app_ms.get("/read/",response_model=list[schemas.read_user]) def read_data(skip: int = 0,limit: int = 10,db: Session = Depends(get_db)): #从数据库中获取用户数据 users = crud.get_user(db, skip=skip,limit=limit) #返回用户数据 return users #录入user @app_ms.post("/create/",response_model = schemas.read_user) def create_data(user: schemas.create_user,db: Session = Depends(get_db)): #检查数据库中是否存在该用户账号 db_user = crud.get_user_by_account(db,name = user.name) #如果存在,则抛出异常 if db_user: raise HTTPException(status_code=400,detail="user already registered") #检查数据库中是否存在该用户手机号 # db_user = crud.get_user_by_phone_number(db,phone_number = user.phone_number) #如果存在,则抛出异常 # if db_user: # raise HTTPException(status_code=400,detail="phone number already used") #返回创建的用户 return crud.create_user(db=db,user=user)from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware import uvicorn from main import app_ms #创建Fastapi应用 app = FastAPI() #允许所有来源的跨域请求 app.add_middleware( CORSMiddleware, allow_origins = ["*"], allow_credentials = True, allow_methods = ["*"], allow_headers = ["*"] ) #包含app_ms路由,前缀为/app_ms,标签为api_ms app.include_router(app_ms,prefix="/api",tags=["api_ms"]) if __name__ == "__main__": #启动uvicorn服务,参数为run:app,主机地址为127.0.0.1,端口号为8000,自动重新加载,工作进程数为1 uvicorn.run(app='run:app',host="127.0.0.1",port=8000,reload=True,workers=1)python run.py 或 uvicorn run:app --reload127.0.0.1:8000/docsuvicorn main:app --reloadpip install pyinstallerpyinstaller --onefile --name myapp run.py