-
guest说:
我要做一个纯后端的API接口项目,文章信息管理系统,用到的框架是FastAPI,使用的模块有sqlalchemy、pydantic、logging,请帮我列出合理的目录结构及文件,给出目录和文件的注释,并给出每个文件内的代码编写,最后打包供下载下来
部署上线。
目录结构:
- config/ # 配置文件目录
- config.py # 配置文件,存放数据库和其他环境配置信息
- doc/ # 文档目录
- README.md # 项目概述文档
- logs/ # 日志目录
- access.log # 访问日志
- error.log # 错误日志
- src/ # 主模块目录
- init.py # 模块初始化文件
- app.py # 启动入口文件
- models.py # 数据模型文件
- routes/ # 路由文件目录
- init.py # 路由文件初始化文件
- articles.py # 文章路由文件
- utils/ # 工具函数目录
- init.py # 工具函数初始化文件
- db.py # 操作数据库工具函数
- requirements.txt # 依赖库文件
- .gitignore # git忽略文件
# 配置文件,存放数据库和其他环境配置信息 import os # 数据库配置 DB_USERNAME = os.getenv("DB_USERNAME") DB_PASSWORD = os.getenv("DB_PASSWORD") DB_HOST = os.getenv("DB_HOST") DB_PORT = os.getenv("DB_PORT") DB_NAME = os.getenv("DB_NAME")
# 启动入口文件 import uvicorn from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from src.models import Base from src.routes.articles import router from src.utils.db import init_db # 初始化app app = FastAPI( title="文章信息管理系统", description="纯后端的API接口项目,用于管理文章信息", version="1.0.0", ) # 跨域设置 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 注册路由 app.include_router(router, prefix="/api/v1") # 初始化数据库 init_db(Base) # 启动服务 if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)
# 数据模型文件 from sqlalchemy import Column, Integer, String from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class Article(Base): __tablename__ = "articles" id = Column(Integer, primary_key=True, index=True) title = Column(String) content = Column(String)
routes/articles.py
# 文章路由文件 from fastapi import APIRouter, Depends from sqlalchemy.orm import Session from src.models import Article from src.utils.db import get_db router = APIRouter() @router.get("/") def get_articles(db: Session = Depends(get_db)): articles = db.query(Article).order_by(Article.id).all() return articles @router.post("/") def create_article(article: Article, db: Session = Depends(get_db)): db.add(article) db.commit() db.refresh(article) return article @router.put("/{article_id}") def update_article(article_id: int, article: Article, db: Session = Depends(get_db)): db.query(Article).filter(Article.id == article_id).update( { "title": article.title, "content": article.content, } ) db.commit() return article @router.delete("/{article_id}") def delete_article(article_id: int, db: Session = Depends(get_db)): article = db.query(Article).get(article_id) db.delete(article) db.commit() return {"msg": "success"}
utils/db.py
# 操作数据库工具函数 from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from src.config import DB_USERNAME, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME # 数据库连接 engine = create_engine( f"mysql+pymysql://{DB_USERNAME}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}?charset=utf8mb4" ) # 数据库Session SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 初始化数据库 def init_db(Base): Base.metadata.create_all(bind=engine) # 获取数据库Session def get_db(): db = SessionLocal() try: yield db finally: db.close()
requirements.txt
fastapi>=0.61.0 uvicorn>=0.11.2 SQLAlchemy>=1.3.16 pydantic>=1.6.1 PyMySQL>=0.9.3
.gitignore
# 日志文件 logs/*
- config/ # 配置文件目录
-
guest说:
我要做一个纯后端的API接口项目,学生信息管理,用的是FastAPI框架,使用的模块有sqlalchemy+pydantic+logging,请帮我列出合理的目录结构及代码文件,并给出每个文件详细的代码编写
内容。
项目目录结构:
├── README.md
├── main.py
├── models
│ ├── init.py
│ ├── base.py
│ ├── student.py
│ └── teacher.py
├── routes
│ ├── init.py
│ ├── student_route.py
│ └── teacher_route.py
└── utils
├── init.py
├── logger.py
├── settings.py
└── validator.pyfrom fastapi import FastAPI from routes import student_route, teacher_route from utils import logger app = FastAPI() app.include_router(student_route.router) app.include_router(teacher_route.router) if __name__ == '__main__': logger.info('Starting FastAPI application...') app.run()
models/base.py:
from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from utils.settings import DATABASE engine = create_engine(DATABASE) Base = declarative_base() Session = sessionmaker(bind=engine) session = Session()
models/student.py:
from sqlalchemy import Column, Integer, String from models.base import Base class Student(Base): __tablename__ = 'students' id = Column(Integer, primary_key=True) name = Column(String(50)) age = Column(Integer)
models/teacher.py:
from sqlalchemy import Column, Integer, String from models.base import Base class Teacher(Base): __tablename__ = 'teachers' id = Column(Integer, primary_key=True) name = Column(String(50)) age = Column(Integer)
routes/student_route.py:
from fastapi import APIRouter from models.student import Student router = APIRouter() @router.get('/students') def get_students(): students = Student.query.all() return students @router.post('/students') def create_student(student: Student): student.save() return student @router.put('/students/{id}') def update_student(id: int, student: Student): student.update(id) return student @router.delete('/students/{id}') def delete_student(id: int): Student.delete(id) return {'message': 'Student deleted successfully'}
routes/teacher_route.py:
from fastapi import APIRouter from models.teacher import Teacher router = APIRouter() @router.get('/teachers') def get_teachers(): teachers = Teacher.query.all() return teachers @router.post('/teachers') def create_teacher(teacher: Teacher): teacher.save() return teacher @router.put('/teachers/{id}') def update_teacher(id: int, teacher: Teacher): teacher.update(id) return teacher @router.delete('/teachers/{id}') def delete_teacher(id: int): Teacher.delete(id) return {'message': 'Teacher deleted successfully'}
utils/logger.py:
import logging logging.basicConfig( level=logging.INFO, format="[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s", datefmt="%H:%M:%S", ) def get_logger(): logger = logging.getLogger(__name__) return logger logger = get_logger()
utils/settings.py:
DATABASE = 'postgresql://postgres:password@localhost:5432/test'
utils/validator.py:
from pydantic import BaseModel, validator class Student(BaseModel): name: str age: int @validator('name') def name_must_contain_space(cls, value): assert ' ' in value, 'name should have a space' return value class Teacher(BaseModel): name: str age: int