以下是一个基于FastAPI框架设计的日志模块，结合SQLite数据库实现增删改查功能的完整实现方案：

1. 项目结构

your_project/
├── app/
│   ├── logs/               # 日志模块目录
│   │   ├── models.py       # 数据库模型定义
│   │   ├── schemas.py      # Pydantic数据验证模型
│   │   ├── crud.py         # 数据库操作逻辑
│   │   └── routes.py       # API路由定义
│   ├── database.py        # 数据库连接配置
│   └── main.py            # FastAPI主入口
├── logs.db                 # SQLite数据库文件
└── requirements.txt

2. 核心代码实现

2.1 数据库配置 (app/database.py)
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmakerSQLALCHEMY_DATABASE_URL = "sqlite:///./logs.db"engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)Base = declarative_base()def get_db():db = SessionLocal()try:yield dbfinally:db.close()
2.2 数据模型 (app/logs/models.py)
from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime
from database import Baseclass Log(Base):__tablename__ = "logs"id = Column(Integer, primary_key=True, index=True)level = Column(String(10), index=True)  # 日志级别 (INFO, WARNING, ERROR)message = Column(String(500))timestamp = Column(DateTime, default=datetime.utcnow)source = Column(String(50), default="application")  # 日志来源
2.3 Pydantic模型 (app/logs/schemas.py)
from pydantic import BaseModel
from datetime import datetimeclass LogCreate(BaseModel):level: strmessage: strsource: str = "application"class LogResponse(LogCreate):id: inttimestamp: datetimeclass Config:orm_mode = True
2.4 CRUD操作 (app/logs/crud.py)
from sqlalchemy.orm import Session
from . import models, schemasdef create_log(db: Session, log: schemas.LogCreate):db_log = models.Log(**log.dict())db.add(db_log)db.commit()db.refresh(db_log)return db_logdef get_logs(db: Session, skip: int = 0, limit: int = 100):return db.query(models.Log).offset(skip).limit(limit).all()def get_log_by_id(db: Session, log_id: int):return db.query(models.Log).filter(models.Log.id == log_id).first()def update_log(db: Session, log_id: int, log: schemas.LogCreate):db_log = db.query(models.Log).filter(models.Log.id == log_id).first()if db_log:for key, value in log.dict().items():setattr(db_log, key, value)db.commit()db.refresh(db_log)return db_logdef delete_log(db: Session, log_id: int):db_log = db.query(models.Log).filter(models.Log.id == log_id).first()if db_log:db.delete(db_log)db.commit()return Truereturn False
2.5 API路由 (app/logs/routes.py)
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from ..database import get_db
from . import schemas, crudrouter = APIRouter(prefix="/logs", tags=["logs"])@router.post("/", response_model=schemas.LogResponse)
def create_log_entry(log: schemas.LogCreate, db: Session = Depends(get_db)):return crud.create_log(db, log)@router.get("/", response_model=list[schemas.LogResponse])
def read_logs(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):return crud.get_logs(db, skip=skip, limit=limit)@router.get("/{log_id}", response_model=schemas.LogResponse)
def read_log(log_id: int, db: Session = Depends(get_db)):db_log = crud.get_log_by_id(db, log_id)if not db_log:raise HTTPException(status_code=404, detail="Log not found")return db_log@router.put("/{log_id}", response_model=schemas.LogResponse)
def update_log_entry(log_id: int, log: schemas.LogCreate, db: Session = Depends(get_db)):updated_log = crud.update_log(db, log_id, log)if not updated_log:raise HTTPException(status_code=404, detail="Log not found")return updated_log@router.delete("/{log_id}")
def delete_log_entry(log_id: int, db: Session = Depends(get_db)):success = crud.delete_log(db, log_id)if not success:raise HTTPException(status_code=404, detail="Log not found")return {"message": "Log deleted successfully"}
2.6 主入口 (app/main.py)
from fastapi import FastAPI
from .database import engine
from .logs.models import Log
from .logs.routes import router as logs_routerLog.metadata.create_all(bind=engine)app = FastAPI()
app.include_router(logs_router)@app.get("/")
def root():return {"message": "Logging System API"}

3. 使用方式

3.1 启动服务
uvicorn app.main:app --reload
3.2 API测试示例
# 创建日志
curl -X POST "http://localhost:8000/logs/" \
-H "Content-Type: application/json" \
-d '{"level": "ERROR", "message": "Database connection failed"}'# 查询所有日志
curl "http://localhost:8000/logs/"# 更新日志
curl -X PUT "http://localhost:8000/logs/1" \
-H "Content-Type: application/json" \
-d '{"level": "WARNING", "message": "Connection timeout"}'# 删除日志
curl -X DELETE "http://localhost:8000/logs/1"

4. 高级功能扩展建议

  1. 日志过滤：
# 在crud.py中添加
def filter_logs(db: Session, level: str = None, source: str = None):query = db.query(models.Log)if level:query = query.filter(models.Log.level == level)if source:query = query.filter(models.Log.source == source)return query.all()
  1. 自动记录请求日志：
# 在main.py中添加中间件
@app.middleware("http")
async def log_requests(request: Request, call_next):start_time = time.time()response = await call_next(request)process_time = (time.time() - start_time) * 1000log_data = {"level": "INFO","message": f"{request.method} {request.url} - {response.status_code}","source": "http"}# 异步写入日志（需配置数据库会话）with SessionLocal() as db:crud.create_log(db, schemas.LogCreate(**log_data))return response
  1. 日志分页查询：
# 在routes.py中改进GET方法
@router.get("/", response_model=list[schemas.LogResponse])
def read_logs(page: int = 1,per_page: int = 20,db: Session = Depends(get_db)
):skip = (page - 1) * per_pagereturn crud.get_logs(db, skip=skip, limit=per_page)

该设计实现了完整的日志管理功能，同时保持了FastAPI的异步特性优势。通过SQLAlchemy ORM层，可以轻松切换其他数据库（如PostgreSQL/MySQL）。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如若转载，请注明出处：http://www.pswp.cn/diannao/82757.shtml
繁体地址，请注明出处：http://hk.pswp.cn/diannao/82757.shtml
英文地址，请注明出处：http://en.pswp.cn/diannao/82757.shtml

如若内容造成侵权/违法违规/事实不符，请联系英文站点网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

面试题：C++虚函数可以是内联函数吗？

面试题：C++虚函数可以是内联函数吗？

目录 1.引言 2.示例 3.总结 1.引言 为什么C的虚函数和内联函数这两个看似矛盾的特性能否共存？这个问题实际上触及了C编译期优化与运行时多态性之间的微妙平衡。我发现这个问题不仅是面试中的常见陷阱，更是理解C深层机制很好的一个点。 虚函数可以被声…
阅读更多...
用python清除PDF文件中的水印（Adobe Acrobat 无法删除）

用python清除PDF文件中的水印（Adobe Acrobat 无法删除）

学校老师发的资料，有时候会带水印，有点强迫症的都想给它去掉。用Adobe Acrobat试了下，检测不到水印，无法删除！分析发现原来这类PDF文件是用word编辑的，其中的水印是加在了页眉中！ 自己动手想办法…
阅读更多...
Oracle中自定义异常内置异常嵌套异常的捕获处理

Oracle中自定义异常内置异常嵌套异常的捕获处理

一、异常类型分类 类型说明示例内置预定义异常Oracle已命名异常（如NO_DATA_FOUND）查询无数据时触发内置非预定义异常未命名的Oracle错误（需用PRAGMA EXCEPTION_INIT关联）ORA-02290（违反检查约束）自定义异常…
阅读更多...
《Flutter社交应用暗黑奥秘：模式适配与色彩的艺术》

《Flutter社交应用暗黑奥秘：模式适配与色彩的艺术》

暗黑模式已从一种新奇的功能演变为用户体验中不可或缺的一环。对于Flutter开发者而言，如何在社交应用中完美实现暗黑模式适配与色彩对比度优化，是一场充满挑战与惊喜的技术探索之旅。 暗黑模式，绝非仅仅是将界面颜色反转这么简单。从用户体验…
阅读更多...
【kubernetes】通过Sealos 命令行工具一键部署k8s集群

【kubernetes】通过Sealos 命令行工具一键部署k8s集群

一、前言 1、sealos安装k8s集群官网：K8s > Quick-start > Deploy-kubernetes | Sealos Docs 2、本文安装的k8s版本为v1.28.9 3、以下是一些基本的安装要求： 每个集群节点应该有不同的主机名。主机名不要带下划线。所有节点的时间需要同步。需要…
阅读更多...
视觉-语言-动作模型：概念、进展、应用与挑战（上）

视觉-语言-动作模型：概念、进展、应用与挑战（上）

25年5月来自 Cornell 大学、香港科大和希腊 U Peloponnese 的论文“Vision-Language-Action Models: Concepts, Progress, Applications and Challenges”。 视觉-语言-动作 (VLA) 模型标志着人工智能的变革性进步，旨在将感知、自然语言理解和具体动作统一在一个计…
阅读更多...
Java笔记4

Java笔记4

第一章 static关键字 2.1 概述 以前我们定义过如下类： public class Student {// 成员变量public String name;public char sex; // 男 女public int age;// 无参数构造方法public Student() {}// 有参数构造方法public Student(String a) {} }我们已经知道面向…
阅读更多...
记一次redis未授权被种挖矿

记一次redis未授权被种挖矿

#挖矿程序 /etc/httpgd /etc/nnt.sh #大小问 #定时任务名为root /var/spool/cron/root 内容：*/50 * * * * sh /etc/nnt.sh >/dev/null 2>&1 定时任务只有所有者可以写，且chmod修改权限失败。 #先查看定时任务的拓展属性，不可变(i…
阅读更多...
Excel分组计算求和的两种实现方案

Excel分组计算求和的两种实现方案

文章目录 背景样例数据方案一、函数求和实现步骤缺点 方案二、数据透视表实现步骤优点 背景 在Excel文档中，经常会进行数据的求和计算，可使用不同的方式实现，记录下来，方便备查。 样例数据 已有商品销量信息，包含销…
阅读更多...
如何应对网站被爬虫和采集？综合防护策略与实用方案

如何应对网站被爬虫和采集？综合防护策略与实用方案

在互联网时代，网站内容被恶意爬虫或采集工具窃取已成为常见问题。这不仅侵犯原创权益，还可能影响网站性能和SEO排名。以下是结合技术、策略与法律的综合解决方案，帮助网站构建有效防护体系。 一、技术防护：阻断爬虫的“技术防线”…
阅读更多...
网卡网孔速率的协商是如何进行的？

网卡网孔速率的协商是如何进行的？

网卡与交换机等网络设备之间的速率协商主要通过**自动协商（Auto-Negotiation）**机制实现，其核心是物理层（PHY）芯片之间的信息交互。以下是协商过程的详细解析： 一、自动协商的核心流程 1. 发送配置帧&am…
阅读更多...
FastExcel 本地开发和Linux上上传Resource文件的差异性

FastExcel 本地开发和Linux上上传Resource文件的差异性

不能直接通过路径来获取 这个是一个下载导出文件的操作 GetMapping(value "/export/all") public void exportAll(HttpServletResponse response, LaylineListReq req) throws IOException {// 从类路径下获取 Excel 文件资源ClassPathResource classPathResource…
阅读更多...
【RAG】Milvus、Pinecone、PgVector向量数据库索引参数优化

【RAG】Milvus、Pinecone、PgVector向量数据库索引参数优化

Milvus 、PgVector 索引参数优化 IVF类索引关键参数（基于聚类算法） nlist (倒排列表数量): 决定将向量空间划分为多少个聚类中心值越大搜索越精确但耗时越长推荐值: 通常设置为数据量的4√n到n/1000之间例如: 1百万数据量可设nlist1000到4000 nprobe (搜…
阅读更多...
5月12日信息差

5月12日信息差

一、国际政治与安全：俄乌冲突与中美博弈 1. 乌克兰战场信息分化 俄方战报： 俄罗斯国防部宣称在顿巴斯地区摧毁乌军12辆坦克及3套美制“海马斯”火箭系统，称乌军反攻受阻。 信息特点：强调装备摧毁数量，淡化前线实际控制变化。 乌方通报： 乌克兰总参谋部表示已夺回巴赫穆特…
阅读更多...
Python如何使用进行风险管理和投资组合优化

Python如何使用进行风险管理和投资组合优化

文章目录 前言python3.13 环境配置风险管理投资组合优化 前言 在 Python 中，可以使用多个库来进行风险管理和投资组合优化，以下是一些常见的方法和库。 python3.13 环境配置 python3.13安装教程：https://blog.csdn.net/2501_91538706/artic…
阅读更多...
C++ 状态模式详解

C++ 状态模式详解

状态模式（State Pattern）是一种行为设计模式，它允许一个对象在内部状态改变时改变其行为，使对象看起来像是改变了其类。 核心概念 设计原则 状态模式遵循以下设计原则： 单一职责原则：将状态相关行为分离…
阅读更多...
Html5新特性_js 给元素自定义属性_json 详解_浅克隆与深克隆

Html5新特性_js 给元素自定义属性_json 详解_浅克隆与深克隆

文章目录 1. html5新特性2.用 js 给元素自定义属性3.json3.1 json与普通对象的区别3.2 json对象与 js对象的转化 4.浅克隆和深克隆 1. html5新特性 html5中引入了新的特性（新的标签），下面的新标签是新的结构标签，不过不太常用 h…
阅读更多...
std::move 和 std::forward

std::move 和 std::forward

关联点 都是执行转换(cast)的函数（函数模板），不产生任何可执行代码。且都可以把实参转换成右值。 std::move无条件将实参（const除外 ）转换成右值引用，std::forward 条件返回右值引用 _EXPORT_STD template…
阅读更多...
Uniapp编写微信小程序，使用canvas进行绘图

Uniapp编写微信小程序，使用canvas进行绘图

一、canvas文档： https://developer.mozilla.org/zh-CN/docs/Web/API/Canvas_API/Tutorial 二、数据绘制（单位是像素）： 1、绘制文本： 文字的长度超过设置的最大宽度，文字会缩在一起 ① 填充文本&#xf…
阅读更多...
FLASH闪存（擦除、编译）

FLASH闪存（擦除、编译）

FLASH闪存 文章目录 FLASH闪存1.存储器映像位置2.FLASH简介3.闪存模块组织3.2闪存的共性： 4.FLASH基本结构4.1FLASH解锁4.2使用指针访问寄存器 5.选项字节5.1选项字节编程5.2选项字节擦除 6.相关函数介绍7.读取内部FLASH（实操）7.1接线图7.2工…
阅读更多...
最新文章

Copyright @ 2022~2023