mirror of
https://github.com/fastapi/fastapi.git
synced 2025-12-26 07:40:57 -05:00
e329d78f86
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
39 lines
937 B
Python
39 lines
937 B
Python
import time
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, FastAPI, HTTPException
|
|
from fastapi.responses import StreamingResponse
|
|
from sqlmodel import Field, Session, SQLModel, create_engine
|
|
|
|
engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")
|
|
|
|
|
|
class User(SQLModel, table=True):
|
|
id: int | None = Field(default=None, primary_key=True)
|
|
name: str
|
|
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
def get_session():
|
|
with Session(engine) as session:
|
|
yield session
|
|
|
|
|
|
def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
|
|
user = session.get(User, user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=403, detail="Not authorized")
|
|
|
|
|
|
def generate_stream(query: str):
|
|
for ch in query:
|
|
yield ch
|
|
time.sleep(0.1)
|
|
|
|
|
|
@app.get("/generate", dependencies=[Depends(get_user)])
|
|
def generate(query: str):
|
|
return StreamingResponse(content=generate_stream(query))
|