from fastapi import FastAPI

# Création de l'instance FastAPI
app = FastAPI()

@app.get("/")
def read_root():
  return {"message": "Hello World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
  return {"item_id": item_id, "q": q}

uvicorn main:app --reload

from typing import List, Dict, Optional
from datetime import date

# Déclaration de types pour les paramètres
def get_user_info(user_id: int, name: str, active: bool = True):
  return {"user_id": user_id, "name": name, "active": active}

from pydantic import BaseModel, EmailStr, Field
from datetime import date
from typing import Optional

class User(BaseModel):
  id: int
  name: str = Field(..., min_length=1, max_length=100)
  email: EmailStr
  joined: date
  is_active: bool = True
  bio: Optional[str] = None

# Validation automatique
user = User(
  id=1,
  name="Alice",
  email="alice@example.com",
  joined="2025-01-15"
)

print(user.dict())

from pydantic import BaseModel
from typing import Optional

class Item(BaseModel):
  name: str
  description: Optional[str] = None
  price: float
  tax: Optional[float] = None

class ItemUpdate(BaseModel):
  name: Optional[str] = None
  description: Optional[str] = None
  price: Optional[float] = None
  tax: Optional[float] = None

from fastapi import FastAPI, HTTPException

app = FastAPI()

# Base de données simulée
items_db = {}

@app.post("/items/", status_code=201)
def create_item(item: Item):
  item_id = len(items_db) + 1
  items_db[item_id] = item
  return {"id": item_id, **item.dict()}

@app.get("/items/{item_id}")
def read_item(item_id: int):
  if item_id not in items_db:
      raise HTTPException(status_code=404, detail="Item not found")
  return items_db[item_id]

@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
  if item_id not in items_db:
      raise HTTPException(status_code=404, detail="Item not found")
  items_db[item_id] = item
  return {"id": item_id, **item.dict()}

@app.patch("/items/{item_id}")
def partial_update_item(item_id: int, item: ItemUpdate):
  if item_id not in items_db:
      raise HTTPException(status_code=404, detail="Item not found")
  
  stored_item = items_db[item_id]
  update_data = item.dict(exclude_unset=True)
  updated_item = stored_item.copy(update=update_data)
  items_db[item_id] = updated_item
  return updated_item

@app.delete("/items/{item_id}", status_code=204)
def delete_item(item_id: int):
  if item_id not in items_db:
      raise HTTPException(status_code=404, detail="Item not found")
  del items_db[item_id]
  return None

from typing import Optional
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
def read_items(
  skip: int = 0,
  limit: int = Query(default=10, le=100),
  q: Optional[str] = Query(None, min_length=3, max_length=50),
  tags: list[str] = Query(default=[])
):
  return {
      "skip": skip,
      "limit": limit,
      "q": q,
      "tags": tags
  }

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBasic, HTTPBasicCredentials

app = FastAPI()
security = HTTPBasic()

def verify_credentials(credentials: HTTPBasicCredentials = Depends(security)):
  correct_username = "admin"
  correct_password = "secret"
  
  if credentials.username != correct_username or credentials.password != correct_password:
      raise HTTPException(status_code=401, detail="Invalid credentials")
  return credentials.username

@app.get("/protected")
def protected_route(username: str = Depends(verify_credentials)):
  return {"message": f"Hello {username}"}

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from datetime import datetime, timedelta

app = FastAPI()

SECRET_KEY = "votre-clé-secrète"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def create_access_token(data: dict, expires_delta: timedelta = None):
  to_encode = data.copy()
  expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
  to_encode.update({"exp": expire})
  return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/token")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
  # Vérification des credentials (simplifié)
  if form_data.username != "user" or form_data.password != "pass":
      raise HTTPException(status_code=401, detail="Incorrect username or password")
  
  access_token = create_access_token(data={"sub": form_data.username})
  return {"access_token": access_token, "token_type": "bearer"}

async def get_current_user(token: str = Depends(oauth2_scheme)):
  try:
      payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
      username: str = payload.get("sub")
      if username is None:
          raise HTTPException(status_code=401, detail="Invalid token")
      return username
  except JWTError:
      raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/users/me")
def read_users_me(current_user: str = Depends(get_current_user)):
  return {"username": current_user}

from fastapi import FastAPI, Depends

app = FastAPI()

def get_db():
  db = "Connection to database"
  try:
      yield db
  finally:
      print("Database connection closed")

@app.get("/users/")
def read_users(db: str = Depends(get_db)):
  return {"db": db, "users": ["Alice", "Bob"]}

from fastapi import FastAPI, Depends, Header, HTTPException

app = FastAPI()

def verify_api_key(x_api_key: str = Header(...)):
  if x_api_key != "secret-key":
      raise HTTPException(status_code=401, detail="Invalid API Key")
  return x_api_key

def get_db():
  return {"connection": "active"}

def get_current_user(
  api_key: str = Depends(verify_api_key),
  db: dict = Depends(get_db)
):
  return {"user": "admin", "authenticated": True}

@app.get("/dashboard")
def dashboard(user: dict = Depends(get_current_user)):
  return {"message": f"Welcome {user['user']}"}

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from fastapi import FastAPI, Depends

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Modèle SQLAlchemy
class UserModel(Base):
  __tablename__ = "users"
  
  id = Column(Integer, primary_key=True, index=True)
  email = Column(String, unique=True, index=True)
  name = Column(String)

Base.metadata.create_all(bind=engine)

# Modèle Pydantic
from pydantic import BaseModel

class UserCreate(BaseModel):
  email: str
  name: str

class User(UserCreate):
  id: int
  
  class Config:
      orm_mode = True

# Dependency
def get_db():
  db = SessionLocal()
  try:
      yield db
  finally:
      db.close()

app = FastAPI()

@app.post("/users/", response_model=User)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
  db_user = UserModel(**user.dict())
  db.add(db_user)
  db.commit()
  db.refresh(db_user)
  return db_user

@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, db: Session = Depends(get_db)):
  user = db.query(UserModel).filter(UserModel.id == user_id).first()
  if user is None:
      raise HTTPException(status_code=404, detail="User not found")
  return user

from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
  <head>
      <title>Chat</title>
  </head>
  <body>
      <h1>WebSocket Chat</h1>
      <ul id='messages'></ul>
      <input id='messageText' autocomplete="off"/>
      <button onclick="sendMessage()">Send</button>
      <script>
          var ws = new WebSocket("ws://localhost:8000/ws");
          ws.onmessage = function(event) {
              var messages = document.getElementById('messages')
              var message = document.createElement('li')
              message.textContent = event.data
              messages.appendChild(message)
          };
          function sendMessage() {
              var input = document.getElementById("messageText")
              ws.send(input.value)
              input.value = ''
          }
      </script>
  </body>
</html>
"""

@app.get("/")
async def get():
  return HTMLResponse(html)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
  await websocket.accept()
  while True:
      data = await websocket.receive_text()
      await websocket.send_text(f"Message: {data}")

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_log(message: str):
  with open("log.txt", "a") as log:
      log.write(f"{message}\n")

@app.post("/send-notification/")
async def send_notification(email: str, background_tasks: BackgroundTasks):
  background_tasks.add_task(write_log, f"Notification sent to {email}")
  return {"message": "Notification sent in background"}

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# CORS
app.add_middleware(
  CORSMiddleware,
  allow_origins=["*"],
  allow_credentials=True,
  allow_methods=["*"],
  allow_headers=["*"],
)

# GZip compression
app.add_middleware(GZipMiddleware, minimum_size=1000)

# Fichiers statiques
app.mount("/static", StaticFiles(directory="static"), name="static")

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.get("/")
def read_root():
  return {"message": "Hello World"}

@app.post("/items/")
def create_item(name: str):
  return {"name": name}

# Tests
client = TestClient(app)

def test_read_root():
  response = client.get("/")
  assert response.status_code == 200
  assert response.json() == {"message": "Hello World"}

def test_create_item():
  response = client.post("/items/?name=Laptop")
  assert response.status_code == 200
  assert response.json() == {"name": "Laptop"}

from fastapi import FastAPI
import strawberry
from strawberry.fastapi import GraphQLRouter

@strawberry.type
class User:
  id: int
  name: str
  email: str

@strawberry.type
class Query:
  @strawberry.field
  def user(self, id: int) -> User:
      return User(id=id, name="Alice", email="alice@example.com")

schema = strawberry.Schema(query=Query)
graphql_app = GraphQLRouter(schema)

app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")