mirror of
https://github.com/stardrophere/InsightRadar.git
synced 2026-06-05 23:32:49 +08:00
stardrophere
71 lines
2.0 KiB
Python
71 lines
2.0 KiB
Python
# app/main.py
|
|
from contextlib import asynccontextmanager
|
|
from fastapi import FastAPI, Depends
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
from app.database import engine, get_db
|
|
from app.models.models import Base, InfoSource, SourceType
|
|
|
|
|
|
# ==========================================
|
|
# 1. 生命周期管理：App 启动时自动建表
|
|
# ==========================================
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
# 启动时执行：检查模型，如果表不存在，自动在 SQLite 中建表！
|
|
print("正在初始化数据库表...")
|
|
# ---> 核心修改 2：直接使用 Base，而不是 models.Base <---
|
|
Base.metadata.create_all(bind=engine)
|
|
print("数据库表初始化完成！")
|
|
yield
|
|
|
|
|
|
|
|
# 初始化 FastAPI，挂载生命周期
|
|
app = FastAPI(title="AI 新闻聚合引擎 API", lifespan=lifespan)
|
|
|
|
|
|
# ==========================================
|
|
# 2. 路由 API 定义
|
|
# ==========================================
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "Welcome to AI News Aggregator API"}
|
|
|
|
|
|
@app.get("/hello/{name}")
|
|
async def say_hello(name: str):
|
|
return {"message": f"Hello {name}"}
|
|
|
|
|
|
# --->与数据库交互的真实接口
|
|
|
|
@app.post("/sources/")
|
|
async def create_info_source(name: str, url: str, db: Session = Depends(get_db)):
|
|
"""
|
|
测试接口：向数据库中添加一个新闻信息源
|
|
"""
|
|
# ---> 核心修改 3：直接使用 InfoSource 和 SourceType <---
|
|
new_source = InfoSource(
|
|
source_name=name,
|
|
source_type=SourceType.RSS_FEED, # 默认用 RSS 测试
|
|
home_url=url
|
|
)
|
|
|
|
db.add(new_source)
|
|
db.commit()
|
|
db.refresh(new_source) # 刷新以获取自动生成的 ID
|
|
|
|
return {"message": "创建成功！", "data": new_source}
|
|
|
|
|
|
@app.get("/sources/")
|
|
async def get_all_sources(db: Session = Depends(get_db)):
|
|
"""
|
|
测试接口：查询数据库中所有的信息源
|
|
"""
|
|
# ---> 核心修改 4：直接使用 InfoSource <---
|
|
sources = db.query(InfoSource).all()
|
|
return {"total": len(sources), "data": sources} |