Motov Yurii
4 months ago
committed by
GitHub
GitHub
694 changed files with 5617 additions and 9020 deletions
@ -0,0 +1,61 @@ |
|||
# Advanced Python Types { #advanced-python-types } |
|||
|
|||
Here are some additional ideas that might be useful when working with Python types. |
|||
|
|||
## Using `Union` or `Optional` { #using-union-or-optional } |
|||
|
|||
If your code for some reason can't use `|`, for example if it's not in a type annotation but in something like `response_model=`, instead of using the vertical bar (`|`) you can use `Union` from `typing`. |
|||
|
|||
For example, you could declare that something could be a `str` or `None`: |
|||
|
|||
```python |
|||
from typing import Union |
|||
|
|||
|
|||
def say_hi(name: Union[str, None]): |
|||
print(f"Hi {name}!") |
|||
``` |
|||
|
|||
`typing` also has a shortcut to declare that something could be `None`, with `Optional`. |
|||
|
|||
Here's a tip from my very **subjective** point of view: |
|||
|
|||
* 🚨 Avoid using `Optional[SomeType]` |
|||
* Instead ✨ **use `Union[SomeType, None]`** ✨. |
|||
|
|||
Both are equivalent and underneath they are the same, but I would recommend `Union` instead of `Optional` because the word "**optional**" would seem to imply that the value is optional, and it actually means "it can be `None`", even if it's not optional and is still required. |
|||
|
|||
I think `Union[SomeType, None]` is more explicit about what it means. |
|||
|
|||
It's just about the words and names. But those words can affect how you and your teammates think about the code. |
|||
|
|||
As an example, let's take this function: |
|||
|
|||
```python |
|||
from typing import Optional |
|||
|
|||
|
|||
def say_hi(name: Optional[str]): |
|||
print(f"Hey {name}!") |
|||
``` |
|||
|
|||
The parameter `name` is defined as `Optional[str]`, but it is **not optional**, you cannot call the function without the parameter: |
|||
|
|||
```Python |
|||
say_hi() # Oh, no, this throws an error! 😱 |
|||
``` |
|||
|
|||
The `name` parameter is **still required** (not *optional*) because it doesn't have a default value. Still, `name` accepts `None` as the value: |
|||
|
|||
```Python |
|||
say_hi(name=None) # This works, None is valid 🎉 |
|||
``` |
|||
|
|||
The good news is, in most cases, you will be able to simply use `|` to define unions of types: |
|||
|
|||
```python |
|||
def say_hi(name: str | None): |
|||
print(f"Hey {name}!") |
|||
``` |
|||
|
|||
So, normally you don't have to worry about names like `Optional` and `Union`. 😎 |
|||
@ -0,0 +1,22 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import JSONResponse |
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
id: str |
|||
value: str |
|||
|
|||
|
|||
class Message(BaseModel): |
|||
message: str |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/{item_id}", response_model=Item, responses={404: {"model": Message}}) |
|||
async def read_item(item_id: str): |
|||
if item_id == "foo": |
|||
return {"id": "foo", "value": "there goes my hero"} |
|||
return JSONResponse(status_code=404, content={"message": "Item not found"}) |
|||
@ -1,30 +0,0 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import FastAPI |
|||
from fastapi.responses import FileResponse |
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
id: str |
|||
value: str |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get( |
|||
"/items/{item_id}", |
|||
response_model=Item, |
|||
responses={ |
|||
200: { |
|||
"content": {"image/png": {}}, |
|||
"description": "Return the JSON item or an image.", |
|||
} |
|||
}, |
|||
) |
|||
async def read_item(item_id: str, img: Union[bool, None] = None): |
|||
if img: |
|||
return FileResponse("image.png", media_type="image/png") |
|||
else: |
|||
return {"id": "foo", "value": "there goes my hero"} |
|||
@ -0,0 +1,37 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import JSONResponse |
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
id: str |
|||
value: str |
|||
|
|||
|
|||
class Message(BaseModel): |
|||
message: str |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get( |
|||
"/items/{item_id}", |
|||
response_model=Item, |
|||
responses={ |
|||
404: {"model": Message, "description": "The item was not found"}, |
|||
200: { |
|||
"description": "Item requested by ID", |
|||
"content": { |
|||
"application/json": { |
|||
"example": {"id": "bar", "value": "The bar tenders"} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
) |
|||
async def read_item(item_id: str): |
|||
if item_id == "foo": |
|||
return {"id": "foo", "value": "there goes my hero"} |
|||
else: |
|||
return JSONResponse(status_code=404, content={"message": "Item not found"}) |
|||
@ -1,32 +0,0 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import FastAPI |
|||
from fastapi.responses import FileResponse |
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
id: str |
|||
value: str |
|||
|
|||
|
|||
responses = { |
|||
404: {"description": "Item not found"}, |
|||
302: {"description": "The item was moved"}, |
|||
403: {"description": "Not enough privileges"}, |
|||
} |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get( |
|||
"/items/{item_id}", |
|||
response_model=Item, |
|||
responses={**responses, 200: {"content": {"image/png": {}}}}, |
|||
) |
|||
async def read_item(item_id: str, img: Union[bool, None] = None): |
|||
if img: |
|||
return FileResponse("image.png", media_type="image/png") |
|||
else: |
|||
return {"id": "foo", "value": "there goes my hero"} |
|||
@ -1,25 +0,0 @@ |
|||
from typing import Annotated, Union |
|||
|
|||
from fastapi import Body, FastAPI, status |
|||
from fastapi.responses import JSONResponse |
|||
|
|||
app = FastAPI() |
|||
|
|||
items = {"foo": {"name": "Fighters", "size": 6}, "bar": {"name": "Tenders", "size": 3}} |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def upsert_item( |
|||
item_id: str, |
|||
name: Annotated[Union[str, None], Body()] = None, |
|||
size: Annotated[Union[int, None], Body()] = None, |
|||
): |
|||
if item_id in items: |
|||
item = items[item_id] |
|||
item["name"] = name |
|||
item["size"] = size |
|||
return item |
|||
else: |
|||
item = {"name": name, "size": size} |
|||
items[item_id] = item |
|||
return JSONResponse(status_code=status.HTTP_201_CREATED, content=item) |
|||
@ -1,25 +0,0 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import Body, FastAPI, status |
|||
from fastapi.responses import JSONResponse |
|||
|
|||
app = FastAPI() |
|||
|
|||
items = {"foo": {"name": "Fighters", "size": 6}, "bar": {"name": "Tenders", "size": 3}} |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def upsert_item( |
|||
item_id: str, |
|||
name: Union[str, None] = Body(default=None), |
|||
size: Union[int, None] = Body(default=None), |
|||
): |
|||
if item_id in items: |
|||
item = items[item_id] |
|||
item["name"] = name |
|||
item["size"] = size |
|||
return item |
|||
else: |
|||
item = {"name": name, "size": size} |
|||
items[item_id] = item |
|||
return JSONResponse(status_code=status.HTTP_201_CREATED, content=item) |
|||
@ -0,0 +1,11 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware |
|||
|
|||
app = FastAPI() |
|||
|
|||
app.add_middleware(HTTPSRedirectMiddleware) |
|||
|
|||
|
|||
@app.get("/") |
|||
async def main(): |
|||
return {"message": "Hello World"} |
|||
@ -0,0 +1,13 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.middleware.trustedhost import TrustedHostMiddleware |
|||
|
|||
app = FastAPI() |
|||
|
|||
app.add_middleware( |
|||
TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"] |
|||
) |
|||
|
|||
|
|||
@app.get("/") |
|||
async def main(): |
|||
return {"message": "Hello World"} |
|||
@ -0,0 +1,11 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.middleware.gzip import GZipMiddleware |
|||
|
|||
app = FastAPI() |
|||
|
|||
app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=5) |
|||
|
|||
|
|||
@app.get("/") |
|||
async def main(): |
|||
return "somebigcontent" |
|||
@ -0,0 +1,8 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/") |
|||
async def read_main(): |
|||
return {"msg": "Hello World"} |
|||
@ -0,0 +1,11 @@ |
|||
from fastapi.testclient import TestClient |
|||
|
|||
from .main import app |
|||
|
|||
client = TestClient(app) |
|||
|
|||
|
|||
def test_read_main(): |
|||
response = client.get("/") |
|||
assert response.status_code == 200 |
|||
assert response.json() == {"msg": "Hello World"} |
|||
@ -1,38 +0,0 @@ |
|||
from typing import Annotated, Union |
|||
|
|||
from fastapi import FastAPI, Header, HTTPException |
|||
from pydantic import BaseModel |
|||
|
|||
fake_secret_token = "coneofsilence" |
|||
|
|||
fake_db = { |
|||
"foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"}, |
|||
"bar": {"id": "bar", "title": "Bar", "description": "The bartenders"}, |
|||
} |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
id: str |
|||
title: str |
|||
description: Union[str, None] = None |
|||
|
|||
|
|||
@app.get("/items/{item_id}", response_model=Item) |
|||
async def read_main(item_id: str, x_token: Annotated[str, Header()]): |
|||
if x_token != fake_secret_token: |
|||
raise HTTPException(status_code=400, detail="Invalid X-Token header") |
|||
if item_id not in fake_db: |
|||
raise HTTPException(status_code=404, detail="Item not found") |
|||
return fake_db[item_id] |
|||
|
|||
|
|||
@app.post("/items/") |
|||
async def create_item(item: Item, x_token: Annotated[str, Header()]) -> Item: |
|||
if x_token != fake_secret_token: |
|||
raise HTTPException(status_code=400, detail="Invalid X-Token header") |
|||
if item.id in fake_db: |
|||
raise HTTPException(status_code=409, detail="Item already exists") |
|||
fake_db[item.id] = item.model_dump() |
|||
return item |
|||
@ -1,65 +0,0 @@ |
|||
from fastapi.testclient import TestClient |
|||
|
|||
from .main import app |
|||
|
|||
client = TestClient(app) |
|||
|
|||
|
|||
def test_read_item(): |
|||
response = client.get("/items/foo", headers={"X-Token": "coneofsilence"}) |
|||
assert response.status_code == 200 |
|||
assert response.json() == { |
|||
"id": "foo", |
|||
"title": "Foo", |
|||
"description": "There goes my hero", |
|||
} |
|||
|
|||
|
|||
def test_read_item_bad_token(): |
|||
response = client.get("/items/foo", headers={"X-Token": "hailhydra"}) |
|||
assert response.status_code == 400 |
|||
assert response.json() == {"detail": "Invalid X-Token header"} |
|||
|
|||
|
|||
def test_read_nonexistent_item(): |
|||
response = client.get("/items/baz", headers={"X-Token": "coneofsilence"}) |
|||
assert response.status_code == 404 |
|||
assert response.json() == {"detail": "Item not found"} |
|||
|
|||
|
|||
def test_create_item(): |
|||
response = client.post( |
|||
"/items/", |
|||
headers={"X-Token": "coneofsilence"}, |
|||
json={"id": "foobar", "title": "Foo Bar", "description": "The Foo Barters"}, |
|||
) |
|||
assert response.status_code == 200 |
|||
assert response.json() == { |
|||
"id": "foobar", |
|||
"title": "Foo Bar", |
|||
"description": "The Foo Barters", |
|||
} |
|||
|
|||
|
|||
def test_create_item_bad_token(): |
|||
response = client.post( |
|||
"/items/", |
|||
headers={"X-Token": "hailhydra"}, |
|||
json={"id": "bazz", "title": "Bazz", "description": "Drop the bazz"}, |
|||
) |
|||
assert response.status_code == 400 |
|||
assert response.json() == {"detail": "Invalid X-Token header"} |
|||
|
|||
|
|||
def test_create_existing_item(): |
|||
response = client.post( |
|||
"/items/", |
|||
headers={"X-Token": "coneofsilence"}, |
|||
json={ |
|||
"id": "foo", |
|||
"title": "The Foo ID Stealers", |
|||
"description": "There goes my stealer", |
|||
}, |
|||
) |
|||
assert response.status_code == 409 |
|||
assert response.json() == {"detail": "Item already exists"} |
|||
@ -1,38 +0,0 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import FastAPI, Header, HTTPException |
|||
from pydantic import BaseModel |
|||
|
|||
fake_secret_token = "coneofsilence" |
|||
|
|||
fake_db = { |
|||
"foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"}, |
|||
"bar": {"id": "bar", "title": "Bar", "description": "The bartenders"}, |
|||
} |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
id: str |
|||
title: str |
|||
description: Union[str, None] = None |
|||
|
|||
|
|||
@app.get("/items/{item_id}", response_model=Item) |
|||
async def read_main(item_id: str, x_token: str = Header()): |
|||
if x_token != fake_secret_token: |
|||
raise HTTPException(status_code=400, detail="Invalid X-Token header") |
|||
if item_id not in fake_db: |
|||
raise HTTPException(status_code=404, detail="Item not found") |
|||
return fake_db[item_id] |
|||
|
|||
|
|||
@app.post("/items/") |
|||
async def create_item(item: Item, x_token: str = Header()) -> Item: |
|||
if x_token != fake_secret_token: |
|||
raise HTTPException(status_code=400, detail="Invalid X-Token header") |
|||
if item.id in fake_db: |
|||
raise HTTPException(status_code=409, detail="Item already exists") |
|||
fake_db[item.id] = item.model_dump() |
|||
return item |
|||
@ -1,65 +0,0 @@ |
|||
from fastapi.testclient import TestClient |
|||
|
|||
from .main import app |
|||
|
|||
client = TestClient(app) |
|||
|
|||
|
|||
def test_read_item(): |
|||
response = client.get("/items/foo", headers={"X-Token": "coneofsilence"}) |
|||
assert response.status_code == 200 |
|||
assert response.json() == { |
|||
"id": "foo", |
|||
"title": "Foo", |
|||
"description": "There goes my hero", |
|||
} |
|||
|
|||
|
|||
def test_read_item_bad_token(): |
|||
response = client.get("/items/foo", headers={"X-Token": "hailhydra"}) |
|||
assert response.status_code == 400 |
|||
assert response.json() == {"detail": "Invalid X-Token header"} |
|||
|
|||
|
|||
def test_read_nonexistent_item(): |
|||
response = client.get("/items/baz", headers={"X-Token": "coneofsilence"}) |
|||
assert response.status_code == 404 |
|||
assert response.json() == {"detail": "Item not found"} |
|||
|
|||
|
|||
def test_create_item(): |
|||
response = client.post( |
|||
"/items/", |
|||
headers={"X-Token": "coneofsilence"}, |
|||
json={"id": "foobar", "title": "Foo Bar", "description": "The Foo Barters"}, |
|||
) |
|||
assert response.status_code == 200 |
|||
assert response.json() == { |
|||
"id": "foobar", |
|||
"title": "Foo Bar", |
|||
"description": "The Foo Barters", |
|||
} |
|||
|
|||
|
|||
def test_create_item_bad_token(): |
|||
response = client.post( |
|||
"/items/", |
|||
headers={"X-Token": "hailhydra"}, |
|||
json={"id": "bazz", "title": "Bazz", "description": "Drop the bazz"}, |
|||
) |
|||
assert response.status_code == 400 |
|||
assert response.json() == {"detail": "Invalid X-Token header"} |
|||
|
|||
|
|||
def test_create_existing_item(): |
|||
response = client.post( |
|||
"/items/", |
|||
headers={"X-Token": "coneofsilence"}, |
|||
json={ |
|||
"id": "foo", |
|||
"title": "The Foo ID Stealers", |
|||
"description": "There goes my stealer", |
|||
}, |
|||
) |
|||
assert response.status_code == 409 |
|||
assert response.json() == {"detail": "Item already exists"} |
|||
@ -0,0 +1,18 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.testclient import TestClient |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/") |
|||
async def read_main(): |
|||
return {"msg": "Hello World"} |
|||
|
|||
|
|||
client = TestClient(app) |
|||
|
|||
|
|||
def test_read_main(): |
|||
response = client.get("/") |
|||
assert response.status_code == 200 |
|||
assert response.json() == {"msg": "Hello World"} |
|||
@ -0,0 +1,31 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.testclient import TestClient |
|||
from fastapi.websockets import WebSocket |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/") |
|||
async def read_main(): |
|||
return {"msg": "Hello World"} |
|||
|
|||
|
|||
@app.websocket("/ws") |
|||
async def websocket(websocket: WebSocket): |
|||
await websocket.accept() |
|||
await websocket.send_json({"msg": "Hello WebSocket"}) |
|||
await websocket.close() |
|||
|
|||
|
|||
def test_read_main(): |
|||
client = TestClient(app) |
|||
response = client.get("/") |
|||
assert response.status_code == 200 |
|||
assert response.json() == {"msg": "Hello World"} |
|||
|
|||
|
|||
def test_websocket(): |
|||
client = TestClient(app) |
|||
with client.websocket_connect("/ws") as websocket: |
|||
data = websocket.receive_json() |
|||
assert data == {"msg": "Hello WebSocket"} |
|||
@ -0,0 +1,24 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.testclient import TestClient |
|||
|
|||
app = FastAPI() |
|||
|
|||
items = {} |
|||
|
|||
|
|||
@app.on_event("startup") |
|||
async def startup_event(): |
|||
items["foo"] = {"name": "Fighters"} |
|||
items["bar"] = {"name": "Tenders"} |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
async def read_items(item_id: str): |
|||
return items[item_id] |
|||
|
|||
|
|||
def test_read_items(): |
|||
with TestClient(app) as client: |
|||
response = client.get("/items/foo") |
|||
assert response.status_code == 200 |
|||
assert response.json() == {"name": "Fighters"} |
|||
@ -0,0 +1,43 @@ |
|||
from contextlib import asynccontextmanager |
|||
|
|||
from fastapi import FastAPI |
|||
from fastapi.testclient import TestClient |
|||
|
|||
items = {} |
|||
|
|||
|
|||
@asynccontextmanager |
|||
async def lifespan(app: FastAPI): |
|||
items["foo"] = {"name": "Fighters"} |
|||
items["bar"] = {"name": "Tenders"} |
|||
yield |
|||
# clean up items |
|||
items.clear() |
|||
|
|||
|
|||
app = FastAPI(lifespan=lifespan) |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
async def read_items(item_id: str): |
|||
return items[item_id] |
|||
|
|||
|
|||
def test_read_items(): |
|||
# Before the lifespan starts, "items" is still empty |
|||
assert items == {} |
|||
|
|||
with TestClient(app) as client: |
|||
# Inside the "with TestClient" block, the lifespan starts and items added |
|||
assert items == {"foo": {"name": "Fighters"}, "bar": {"name": "Tenders"}} |
|||
|
|||
response = client.get("/items/foo") |
|||
assert response.status_code == 200 |
|||
assert response.json() == {"name": "Fighters"} |
|||
|
|||
# After the requests is done, the items are still there |
|||
assert items == {"foo": {"name": "Fighters"}, "bar": {"name": "Tenders"}} |
|||
|
|||
# The end of the "with TestClient" block simulates terminating the app, so |
|||
# the lifespan ends and items are cleaned up |
|||
assert items == {} |
|||
@ -0,0 +1,8 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/") |
|||
async def root(): |
|||
return {"message": "Tomato"} |
|||
@ -0,0 +1,14 @@ |
|||
import pytest |
|||
from httpx import ASGITransport, AsyncClient |
|||
|
|||
from .main import app |
|||
|
|||
|
|||
@pytest.mark.anyio |
|||
async def test_root(): |
|||
async with AsyncClient( |
|||
transport=ASGITransport(app=app), base_url="http://test" |
|||
) as ac: |
|||
response = await ac.get("/") |
|||
assert response.status_code == 200 |
|||
assert response.json() == {"message": "Tomato"} |
|||
@ -0,0 +1,21 @@ |
|||
from typing import Annotated |
|||
|
|||
from fastapi import Depends, FastAPI, HTTPException, status |
|||
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class HTTPBearer403(HTTPBearer): |
|||
def make_not_authenticated_error(self) -> HTTPException: |
|||
return HTTPException( |
|||
status_code=status.HTTP_403_FORBIDDEN, detail="Not authenticated" |
|||
) |
|||
|
|||
|
|||
CredentialsDep = Annotated[HTTPAuthorizationCredentials, Depends(HTTPBearer403())] |
|||
|
|||
|
|||
@app.get("/me") |
|||
def read_me(credentials: CredentialsDep): |
|||
return {"message": "You are authenticated", "token": credentials.credentials} |
|||
@ -0,0 +1,15 @@ |
|||
from fastapi import BackgroundTasks, FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
def write_notification(email: str, message=""): |
|||
with open("log.txt", mode="w") as email_file: |
|||
content = f"notification for {email}: {message}" |
|||
email_file.write(content) |
|||
|
|||
|
|||
@app.post("/send-notification/{email}") |
|||
async def send_notification(email: str, background_tasks: BackgroundTasks): |
|||
background_tasks.add_task(write_notification, email, message="some notification") |
|||
return {"message": "Notification sent in the background"} |
|||
@ -1,26 +0,0 @@ |
|||
from typing import Annotated, Union |
|||
|
|||
from fastapi import BackgroundTasks, Depends, FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
def write_log(message: str): |
|||
with open("log.txt", mode="a") as log: |
|||
log.write(message) |
|||
|
|||
|
|||
def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None): |
|||
if q: |
|||
message = f"found query: {q}\n" |
|||
background_tasks.add_task(write_log, message) |
|||
return q |
|||
|
|||
|
|||
@app.post("/send-notification/{email}") |
|||
async def send_notification( |
|||
email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)] |
|||
): |
|||
message = f"message to {email}\n" |
|||
background_tasks.add_task(write_log, message) |
|||
return {"message": "Message sent"} |
|||
@ -1,26 +0,0 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import BackgroundTasks, Depends, FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
def write_log(message: str): |
|||
with open("log.txt", mode="a") as log: |
|||
log.write(message) |
|||
|
|||
|
|||
def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None): |
|||
if q: |
|||
message = f"found query: {q}\n" |
|||
background_tasks.add_task(write_log, message) |
|||
return q |
|||
|
|||
|
|||
@app.post("/send-notification/{email}") |
|||
async def send_notification( |
|||
email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query) |
|||
): |
|||
message = f"message to {email}\n" |
|||
background_tasks.add_task(write_log, message) |
|||
return {"message": "Message sent"} |
|||
@ -0,0 +1,8 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/") |
|||
def read_items(): |
|||
return ["plumbus", "portal gun"] |
|||
@ -0,0 +1,8 @@ |
|||
from fastapi import FastAPI, Request |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/app") |
|||
def read_main(request: Request): |
|||
return {"message": "Hello World", "root_path": request.scope.get("root_path")} |
|||
@ -0,0 +1,8 @@ |
|||
from fastapi import FastAPI, Request |
|||
|
|||
app = FastAPI(root_path="/api/v1") |
|||
|
|||
|
|||
@app.get("/app") |
|||
def read_main(request: Request): |
|||
return {"message": "Hello World", "root_path": request.scope.get("root_path")} |
|||
@ -0,0 +1,14 @@ |
|||
from fastapi import FastAPI, Request |
|||
|
|||
app = FastAPI( |
|||
servers=[ |
|||
{"url": "https://stag.example.com", "description": "Staging environment"}, |
|||
{"url": "https://prod.example.com", "description": "Production environment"}, |
|||
], |
|||
root_path="/api/v1", |
|||
) |
|||
|
|||
|
|||
@app.get("/app") |
|||
def read_main(request: Request): |
|||
return {"message": "Hello World", "root_path": request.scope.get("root_path")} |
|||
@ -0,0 +1,15 @@ |
|||
from fastapi import FastAPI, Request |
|||
|
|||
app = FastAPI( |
|||
servers=[ |
|||
{"url": "https://stag.example.com", "description": "Staging environment"}, |
|||
{"url": "https://prod.example.com", "description": "Production environment"}, |
|||
], |
|||
root_path="/api/v1", |
|||
root_path_in_servers=False, |
|||
) |
|||
|
|||
|
|||
@app.get("/app") |
|||
def read_main(request: Request): |
|||
return {"message": "Hello World", "root_path": request.scope.get("root_path")} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue