41 changed files with 26 additions and 419 deletions
@ -1,25 +0,0 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import Depends, FastAPI |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
async def common_parameters( |
|||
q: Union[str, None] = None, skip: int = 0, limit: int = 100 |
|||
): |
|||
return {"q": q, "skip": skip, "limit": limit} |
|||
|
|||
|
|||
CommonsDep = Annotated[dict, Depends(common_parameters)] |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(commons: CommonsDep): |
|||
return commons |
|||
|
|||
|
|||
@app.get("/users/") |
|||
async def read_users(commons: CommonsDep): |
|||
return commons |
|||
@ -1,22 +0,0 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import Depends, FastAPI |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
async def common_parameters( |
|||
q: Union[str, None] = None, skip: int = 0, limit: int = 100 |
|||
): |
|||
return {"q": q, "skip": skip, "limit": limit} |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(commons: Annotated[dict, Depends(common_parameters)]): |
|||
return commons |
|||
|
|||
|
|||
@app.get("/users/") |
|||
async def read_users(commons: Annotated[dict, Depends(common_parameters)]): |
|||
return commons |
|||
@ -1,26 +0,0 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import Depends, FastAPI |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}] |
|||
|
|||
|
|||
class CommonQueryParams: |
|||
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100): |
|||
self.q = q |
|||
self.skip = skip |
|||
self.limit = limit |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]): |
|||
response = {} |
|||
if commons.q: |
|||
response.update({"q": commons.q}) |
|||
items = fake_items_db[commons.skip : commons.skip + commons.limit] |
|||
response.update({"items": items}) |
|||
return response |
|||
@ -1,26 +0,0 @@ |
|||
from typing import Any, Union |
|||
|
|||
from fastapi import Depends, FastAPI |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}] |
|||
|
|||
|
|||
class CommonQueryParams: |
|||
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100): |
|||
self.q = q |
|||
self.skip = skip |
|||
self.limit = limit |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(commons: Annotated[Any, Depends(CommonQueryParams)]): |
|||
response = {} |
|||
if commons.q: |
|||
response.update({"q": commons.q}) |
|||
items = fake_items_db[commons.skip : commons.skip + commons.limit] |
|||
response.update({"items": items}) |
|||
return response |
|||
@ -1,26 +0,0 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import Depends, FastAPI |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}] |
|||
|
|||
|
|||
class CommonQueryParams: |
|||
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100): |
|||
self.q = q |
|||
self.skip = skip |
|||
self.limit = limit |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(commons: Annotated[CommonQueryParams, Depends()]): |
|||
response = {} |
|||
if commons.q: |
|||
response.update({"q": commons.q}) |
|||
items = fake_items_db[commons.skip : commons.skip + commons.limit] |
|||
response.update({"items": items}) |
|||
return response |
|||
@ -1,26 +0,0 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import Cookie, Depends, FastAPI |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
def query_extractor(q: Union[str, None] = None): |
|||
return q |
|||
|
|||
|
|||
def query_or_cookie_extractor( |
|||
q: Annotated[str, Depends(query_extractor)], |
|||
last_query: Annotated[Union[str, None], Cookie()] = None, |
|||
): |
|||
if not q: |
|||
return last_query |
|||
return q |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_query( |
|||
query_or_default: Annotated[str, Depends(query_or_cookie_extractor)], |
|||
): |
|||
return {"q_or_cookie": query_or_default} |
|||
@ -1,20 +0,0 @@ |
|||
from fastapi import Depends, FastAPI, Header, HTTPException |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
async def verify_token(x_token: Annotated[str, Header()]): |
|||
if x_token != "fake-super-secret-token": |
|||
raise HTTPException(status_code=400, detail="X-Token header invalid") |
|||
|
|||
|
|||
async def verify_key(x_key: Annotated[str, Header()]): |
|||
if x_key != "fake-super-secret-key": |
|||
raise HTTPException(status_code=400, detail="X-Key header invalid") |
|||
return x_key |
|||
|
|||
|
|||
@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)]) |
|||
async def read_items(): |
|||
return [{"item": "Foo"}, {"item": "Bar"}] |
|||
@ -1,26 +0,0 @@ |
|||
from fastapi import Depends |
|||
from typing_extensions import Annotated |
|||
|
|||
|
|||
async def dependency_a(): |
|||
dep_a = generate_dep_a() |
|||
try: |
|||
yield dep_a |
|||
finally: |
|||
dep_a.close() |
|||
|
|||
|
|||
async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]): |
|||
dep_b = generate_dep_b() |
|||
try: |
|||
yield dep_b |
|||
finally: |
|||
dep_b.close(dep_a) |
|||
|
|||
|
|||
async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]): |
|||
dep_c = generate_dep_c() |
|||
try: |
|||
yield dep_c |
|||
finally: |
|||
dep_c.close(dep_b) |
|||
@ -1,31 +0,0 @@ |
|||
from fastapi import Depends, FastAPI, HTTPException |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
data = { |
|||
"plumbus": {"description": "Freshly pickled plumbus", "owner": "Morty"}, |
|||
"portal-gun": {"description": "Gun to create portals", "owner": "Rick"}, |
|||
} |
|||
|
|||
|
|||
class OwnerError(Exception): |
|||
pass |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
except OwnerError as e: |
|||
raise HTTPException(status_code=400, detail=f"Owner error: {e}") |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]): |
|||
if item_id not in data: |
|||
raise HTTPException(status_code=404, detail="Item not found") |
|||
item = data[item_id] |
|||
if item["owner"] != username: |
|||
raise OwnerError(username) |
|||
return item |
|||
@ -1,28 +0,0 @@ |
|||
from fastapi import Depends, FastAPI, HTTPException |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class InternalError(Exception): |
|||
pass |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
except InternalError: |
|||
print("Oops, we didn't raise again, Britney 😱") |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]): |
|||
if item_id == "portal-gun": |
|||
raise InternalError( |
|||
f"The portal gun is too dangerous to be owned by {username}" |
|||
) |
|||
if item_id != "plumbus": |
|||
raise HTTPException( |
|||
status_code=404, detail="Item not found, there's only a plumbus here" |
|||
) |
|||
return item_id |
|||
@ -1,29 +0,0 @@ |
|||
from fastapi import Depends, FastAPI, HTTPException |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class InternalError(Exception): |
|||
pass |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
except InternalError: |
|||
print("We don't swallow the internal error here, we raise again 😎") |
|||
raise |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]): |
|||
if item_id == "portal-gun": |
|||
raise InternalError( |
|||
f"The portal gun is too dangerous to be owned by {username}" |
|||
) |
|||
if item_id != "plumbus": |
|||
raise HTTPException( |
|||
status_code=404, detail="Item not found, there's only a plumbus here" |
|||
) |
|||
return item_id |
|||
@ -1,16 +0,0 @@ |
|||
from fastapi import Depends, FastAPI |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
finally: |
|||
print("Cleanup up before response is sent") |
|||
|
|||
|
|||
@app.get("/users/me") |
|||
def get_user_me(username: Annotated[str, Depends(get_username, scope="function")]): |
|||
return username |
|||
@ -1,25 +0,0 @@ |
|||
from fastapi import Depends |
|||
|
|||
|
|||
async def dependency_a(): |
|||
dep_a = generate_dep_a() |
|||
try: |
|||
yield dep_a |
|||
finally: |
|||
dep_a.close() |
|||
|
|||
|
|||
async def dependency_b(dep_a=Depends(dependency_a)): |
|||
dep_b = generate_dep_b() |
|||
try: |
|||
yield dep_b |
|||
finally: |
|||
dep_b.close(dep_a) |
|||
|
|||
|
|||
async def dependency_c(dep_b=Depends(dependency_b)): |
|||
dep_c = generate_dep_c() |
|||
try: |
|||
yield dep_c |
|||
finally: |
|||
dep_c.close(dep_b) |
|||
@ -1,22 +0,0 @@ |
|||
from fastapi import Depends, FastAPI |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class FixedContentQueryChecker: |
|||
def __init__(self, fixed_content: str): |
|||
self.fixed_content = fixed_content |
|||
|
|||
def __call__(self, q: str = ""): |
|||
if q: |
|||
return self.fixed_content in q |
|||
return False |
|||
|
|||
|
|||
checker = FixedContentQueryChecker("bar") |
|||
|
|||
|
|||
@app.get("/query-checker/") |
|||
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]): |
|||
return {"fixed_content_in_query": fixed_content_included} |
|||
@ -1,26 +0,0 @@ |
|||
from fastapi import Depends, FastAPI, Header, HTTPException |
|||
from typing_extensions import Annotated |
|||
|
|||
|
|||
async def verify_token(x_token: Annotated[str, Header()]): |
|||
if x_token != "fake-super-secret-token": |
|||
raise HTTPException(status_code=400, detail="X-Token header invalid") |
|||
|
|||
|
|||
async def verify_key(x_key: Annotated[str, Header()]): |
|||
if x_key != "fake-super-secret-key": |
|||
raise HTTPException(status_code=400, detail="X-Key header invalid") |
|||
return x_key |
|||
|
|||
|
|||
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)]) |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(): |
|||
return [{"item": "Portal Gun"}, {"item": "Plumbus"}] |
|||
|
|||
|
|||
@app.get("/users/") |
|||
async def read_users(): |
|||
return [{"username": "Rick"}, {"username": "Morty"}] |
|||
Loading…
Reference in new issue