Motov Yurii
7 months ago
committed by
GitHub
GitHub
102 changed files with 3038 additions and 585 deletions
@ -24,7 +24,9 @@ jobs: |
|||||
env: |
env: |
||||
GITHUB_CONTEXT: ${{ toJson(github) }} |
GITHUB_CONTEXT: ${{ toJson(github) }} |
||||
run: echo "$GITHUB_CONTEXT" |
run: echo "$GITHUB_CONTEXT" |
||||
- uses: actions/checkout@v6 |
# pin to actions/checkout@v5 for compatibility with latest-changes |
||||
|
# Ref: https://github.com/actions/checkout/issues/2313 |
||||
|
- uses: actions/checkout@v5 |
||||
with: |
with: |
||||
# To allow latest-changes to commit to the main branch |
# To allow latest-changes to commit to the main branch |
||||
token: ${{ secrets.FASTAPI_LATEST_CHANGES }} |
token: ${{ secrets.FASTAPI_LATEST_CHANGES }} |
||||
@ -34,7 +36,7 @@ jobs: |
|||||
if: ${{ github.event_name == 'workflow_dispatch' && github.event.inputs.debug_enabled == 'true' }} |
if: ${{ github.event_name == 'workflow_dispatch' && github.event.inputs.debug_enabled == 'true' }} |
||||
with: |
with: |
||||
limit-access-to-actor: true |
limit-access-to-actor: true |
||||
- uses: tiangolo/[email protected].0 |
- uses: tiangolo/[email protected].1 |
||||
with: |
with: |
||||
token: ${{ secrets.GITHUB_TOKEN }} |
token: ${{ secrets.GITHUB_TOKEN }} |
||||
latest_changes_file: docs/en/docs/release-notes.md |
latest_changes_file: docs/en/docs/release-notes.md |
||||
|
|||||
@ -1,16 +1,24 @@ |
|||||
# FastAPI bei Cloudanbietern bereitstellen { #deploy-fastapi-on-cloud-providers } |
# FastAPI bei Cloudanbietern deployen { #deploy-fastapi-on-cloud-providers } |
||||
|
|
||||
Sie können praktisch **jeden Cloudanbieter** verwenden, um Ihre FastAPI-Anwendung bereitzustellen. |
Sie können praktisch **jeden Cloudanbieter** verwenden, um Ihre FastAPI-Anwendung bereitzustellen. |
||||
|
|
||||
In den meisten Fällen bieten die großen Cloudanbieter Anleitungen zum Bereitstellen von FastAPI an. |
In den meisten Fällen bieten die großen Cloudanbieter Anleitungen zum Deployment von FastAPI an. |
||||
|
|
||||
## Cloudanbieter – Sponsoren { #cloud-providers-sponsors } |
## FastAPI Cloud { #fastapi-cloud } |
||||
|
|
||||
|
**<a href="https://fastapicloud.com" class="external-link" target="_blank">FastAPI Cloud</a>** wurde vom selben Autor und Team hinter **FastAPI** entwickelt. |
||||
|
|
||||
|
Es vereinfacht den Prozess des **Erstellens**, **Deployens** und **Zugreifens** auf eine API mit minimalem Aufwand. |
||||
|
|
||||
Einige Cloudanbieter ✨ [**sponsern FastAPI**](../help-fastapi.md#sponsor-the-author){.internal-link target=_blank} ✨, dies stellt die kontinuierliche und gesunde **Entwicklung** von FastAPI und seinem **Ökosystem** sicher. |
Es bringt die gleiche **Developer-Experience** beim Erstellen von Apps mit FastAPI auch zum **Deployment** in der Cloud. 🎉 |
||||
|
|
||||
|
FastAPI Cloud ist der Hauptsponsor und Finanzierungsgeber für die *FastAPI and friends* Open-Source-Projekte. ✨ |
||||
|
|
||||
|
## Cloudanbieter – Sponsoren { #cloud-providers-sponsors } |
||||
|
|
||||
Und es zeigt ihr wahres Engagement für FastAPI und seine **Community** (Sie), da sie Ihnen nicht nur einen **guten Service** bieten möchten, sondern auch sicherstellen möchten, dass Sie ein **gutes und gesundes Framework**, FastAPI, haben. 🙇 |
Einige andere Cloudanbieter ✨ [**sponsern FastAPI**](../help-fastapi.md#sponsor-the-author){.internal-link target=_blank} ✨ ebenfalls. 🙇 |
||||
|
|
||||
Vielleicht möchten Sie deren Dienste ausprobieren und deren Anleitungen folgen: |
Sie könnten diese ebenfalls in Betracht ziehen, deren Anleitungen folgen und ihre Dienste ausprobieren: |
||||
|
|
||||
* <a href="https://docs.render.com/deploy-fastapi?utm_source=deploydoc&utm_medium=referral&utm_campaign=fastapi" class="external-link" target="_blank">Render</a> |
* <a href="https://docs.render.com/deploy-fastapi?utm_source=deploydoc&utm_medium=referral&utm_campaign=fastapi" class="external-link" target="_blank">Render</a> |
||||
* <a href="https://docs.railway.com/guides/fastapi?utm_medium=integration&utm_source=docs&utm_campaign=fastapi" class="external-link" target="_blank">Railway</a> |
* <a href="https://docs.railway.com/guides/fastapi?utm_medium=integration&utm_source=docs&utm_campaign=fastapi" class="external-link" target="_blank">Railway</a> |
||||
|
|||||
@ -0,0 +1,65 @@ |
|||||
|
# FastAPI Cloud { #fastapi-cloud } |
||||
|
|
||||
|
Sie können Ihre FastAPI-App in der <a href="https://fastapicloud.com" class="external-link" target="_blank">FastAPI Cloud</a> mit **einem einzigen Befehl** deployen – tragen Sie sich in die Warteliste ein, falls noch nicht geschehen. 🚀 |
||||
|
|
||||
|
## Anmelden { #login } |
||||
|
|
||||
|
Stellen Sie sicher, dass Sie bereits ein **FastAPI-Cloud-Konto** haben (wir haben Sie von der Warteliste eingeladen 😉). |
||||
|
|
||||
|
Melden Sie sich dann an: |
||||
|
|
||||
|
<div class="termy"> |
||||
|
|
||||
|
```console |
||||
|
$ fastapi login |
||||
|
|
||||
|
You are logged in to FastAPI Cloud 🚀 |
||||
|
``` |
||||
|
|
||||
|
</div> |
||||
|
|
||||
|
## Deployen { #deploy } |
||||
|
|
||||
|
Stellen Sie Ihre App jetzt mit **einem einzigen Befehl** bereit: |
||||
|
|
||||
|
<div class="termy"> |
||||
|
|
||||
|
```console |
||||
|
$ fastapi deploy |
||||
|
|
||||
|
Deploying to FastAPI Cloud... |
||||
|
|
||||
|
✅ Deployment successful! |
||||
|
|
||||
|
🐔 Ready the chicken! Your app is ready at https://myapp.fastapicloud.dev |
||||
|
``` |
||||
|
|
||||
|
</div> |
||||
|
|
||||
|
Das war’s! Jetzt können Sie Ihre App unter dieser URL aufrufen. ✨ |
||||
|
|
||||
|
## Über FastAPI Cloud { #about-fastapi-cloud } |
||||
|
|
||||
|
**<a href="https://fastapicloud.com" class="external-link" target="_blank">FastAPI Cloud</a>** wird vom gleichen Autor und Team hinter **FastAPI** entwickelt. |
||||
|
|
||||
|
Es vereinfacht den Prozess des **Erstellens**, **Deployens** und **Nutzens** einer API mit minimalem Aufwand. |
||||
|
|
||||
|
Es bringt die gleiche **Developer-Experience** beim Erstellen von Apps mit FastAPI auch zum **Deployment** in der Cloud. 🎉 |
||||
|
|
||||
|
Es kümmert sich außerdem um das meiste, was beim Deployen einer App nötig ist, zum Beispiel: |
||||
|
|
||||
|
* HTTPS |
||||
|
* Replikation, mit Autoscaling basierend auf Requests |
||||
|
* usw. |
||||
|
|
||||
|
FastAPI Cloud ist Hauptsponsor und Finanzierer der Open-Source-Projekte *FastAPI and friends*. ✨ |
||||
|
|
||||
|
## Bei anderen Cloudanbietern deployen { #deploy-to-other-cloud-providers } |
||||
|
|
||||
|
FastAPI ist Open Source und basiert auf Standards. Sie können FastAPI-Apps bei jedem Cloudanbieter Ihrer Wahl deployen. |
||||
|
|
||||
|
Folgen Sie den Anleitungen Ihres Cloudanbieters, um dort FastAPI-Apps zu deployen. 🤓 |
||||
|
|
||||
|
## Auf den eigenen Server deployen { #deploy-your-own-server } |
||||
|
|
||||
|
Ich werde Ihnen später in diesem **Deployment-Leitfaden** auch alle Details zeigen, sodass Sie verstehen, was passiert, was geschehen muss und wie Sie FastAPI-Apps selbst deployen können, auch auf Ihre eigenen Server. 🤓 |
||||
@ -0,0 +1,17 @@ |
|||||
|
# Alte 403-Authentifizierungsfehler-Statuscodes verwenden { #use-old-403-authentication-error-status-codes } |
||||
|
|
||||
|
Vor FastAPI-Version `0.122.0` verwendeten die integrierten Sicherheits-Utilities den HTTP-Statuscode `403 Forbidden`, wenn sie dem Client nach einer fehlgeschlagenen Authentifizierung einen Fehler zurückgaben. |
||||
|
|
||||
|
Ab FastAPI-Version `0.122.0` verwenden sie den passenderen HTTP-Statuscode `401 Unauthorized` und geben in der Response einen sinnvollen `WWW-Authenticate`-Header zurück, gemäß den HTTP-Spezifikationen, <a href="https://datatracker.ietf.org/doc/html/rfc7235#section-3.1" class="external-link" target="_blank">RFC 7235</a>, <a href="https://datatracker.ietf.org/doc/html/rfc9110#name-401-unauthorized" class="external-link" target="_blank">RFC 9110</a>. |
||||
|
|
||||
|
Aber falls Ihre Clients aus irgendeinem Grund vom alten Verhalten abhängen, können Sie darauf zurückgreifen, indem Sie in Ihren Sicherheitsklassen die Methode `make_not_authenticated_error` überschreiben. |
||||
|
|
||||
|
Sie können beispielsweise eine Unterklasse von `HTTPBearer` erstellen, die einen Fehler `403 Forbidden` zurückgibt, statt des Default-`401 Unauthorized`-Fehlers: |
||||
|
|
||||
|
{* ../../docs_src/authentication_error_status_code/tutorial001_an_py39.py hl[9:13] *} |
||||
|
|
||||
|
/// tip | Tipp |
||||
|
|
||||
|
Beachten Sie, dass die Funktion die Exception-Instanz zurückgibt; sie wirft sie nicht. Das Werfen erfolgt im restlichen internen Code. |
||||
|
|
||||
|
/// |
||||
@ -0,0 +1,17 @@ |
|||||
|
# Use Old 403 Authentication Error Status Codes { #use-old-403-authentication-error-status-codes } |
||||
|
|
||||
|
Before FastAPI version `0.122.0`, when the integrated security utilities returned an error to the client after a failed authentication, they used the HTTP status code `403 Forbidden`. |
||||
|
|
||||
|
Starting with FastAPI version `0.122.0`, they use the more appropriate HTTP status code `401 Unauthorized`, and return a sensible `WWW-Authenticate` header in the response, following the HTTP specifications, <a href="https://datatracker.ietf.org/doc/html/rfc7235#section-3.1" class="external-link" target="_blank">RFC 7235</a>, <a href="https://datatracker.ietf.org/doc/html/rfc9110#name-401-unauthorized" class="external-link" target="_blank">RFC 9110</a>. |
||||
|
|
||||
|
But if for some reason your clients depend on the old behavior, you can revert to it by overriding the method `make_not_authenticated_error` in your security classes. |
||||
|
|
||||
|
For example, you can create a subclass of `HTTPBearer` that returns a `403 Forbidden` error instead of the default `401 Unauthorized` error: |
||||
|
|
||||
|
{* ../../docs_src/authentication_error_status_code/tutorial001_an_py39.py hl[9:13] *} |
||||
|
|
||||
|
/// tip |
||||
|
|
||||
|
Notice that the function returns the exception instance, it doesn't raise it. The raising is done in the rest of the internal code. |
||||
|
|
||||
|
/// |
||||
|
After Width: | Height: | Size: 7.1 KiB |
|
After Width: | Height: | Size: 7.2 KiB |
@ -0,0 +1,20 @@ |
|||||
|
from fastapi import Depends, FastAPI, HTTPException, status |
||||
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer |
||||
|
from typing_extensions import Annotated |
||||
|
|
||||
|
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,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,9 @@ |
|||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
|
||||
|
def forwardref_method(input: "ForwardRefModel") -> "ForwardRefModel": |
||||
|
return ForwardRefModel(x=input.x + 1) |
||||
|
|
||||
|
|
||||
|
class ForwardRefModel(BaseModel): |
||||
|
x: int = 0 |
||||
@ -0,0 +1,251 @@ |
|||||
|
from functools import partial |
||||
|
from typing import AsyncGenerator, Generator |
||||
|
|
||||
|
import pytest |
||||
|
from fastapi import Depends, FastAPI |
||||
|
from fastapi.testclient import TestClient |
||||
|
from typing_extensions import Annotated |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
def function_dependency(value: str) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
async def async_function_dependency(value: str) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
def gen_dependency(value: str) -> Generator[str, None, None]: |
||||
|
yield value |
||||
|
|
||||
|
|
||||
|
async def async_gen_dependency(value: str) -> AsyncGenerator[str, None]: |
||||
|
yield value |
||||
|
|
||||
|
|
||||
|
class CallableDependency: |
||||
|
def __call__(self, value: str) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
class CallableGenDependency: |
||||
|
def __call__(self, value: str) -> Generator[str, None, None]: |
||||
|
yield value |
||||
|
|
||||
|
|
||||
|
class AsyncCallableDependency: |
||||
|
async def __call__(self, value: str) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
class AsyncCallableGenDependency: |
||||
|
async def __call__(self, value: str) -> AsyncGenerator[str, None]: |
||||
|
yield value |
||||
|
|
||||
|
|
||||
|
class MethodsDependency: |
||||
|
def synchronous(self, value: str) -> str: |
||||
|
return value |
||||
|
|
||||
|
async def asynchronous(self, value: str) -> str: |
||||
|
return value |
||||
|
|
||||
|
def synchronous_gen(self, value: str) -> Generator[str, None, None]: |
||||
|
yield value |
||||
|
|
||||
|
async def asynchronous_gen(self, value: str) -> AsyncGenerator[str, None]: |
||||
|
yield value |
||||
|
|
||||
|
|
||||
|
callable_dependency = CallableDependency() |
||||
|
callable_gen_dependency = CallableGenDependency() |
||||
|
async_callable_dependency = AsyncCallableDependency() |
||||
|
async_callable_gen_dependency = AsyncCallableGenDependency() |
||||
|
methods_dependency = MethodsDependency() |
||||
|
|
||||
|
|
||||
|
@app.get("/partial-function-dependency") |
||||
|
async def get_partial_function_dependency( |
||||
|
value: Annotated[ |
||||
|
str, Depends(partial(function_dependency, "partial-function-dependency")) |
||||
|
], |
||||
|
) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/partial-async-function-dependency") |
||||
|
async def get_partial_async_function_dependency( |
||||
|
value: Annotated[ |
||||
|
str, |
||||
|
Depends( |
||||
|
partial(async_function_dependency, "partial-async-function-dependency") |
||||
|
), |
||||
|
], |
||||
|
) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/partial-gen-dependency") |
||||
|
async def get_partial_gen_dependency( |
||||
|
value: Annotated[str, Depends(partial(gen_dependency, "partial-gen-dependency"))], |
||||
|
) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/partial-async-gen-dependency") |
||||
|
async def get_partial_async_gen_dependency( |
||||
|
value: Annotated[ |
||||
|
str, Depends(partial(async_gen_dependency, "partial-async-gen-dependency")) |
||||
|
], |
||||
|
) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/partial-callable-dependency") |
||||
|
async def get_partial_callable_dependency( |
||||
|
value: Annotated[ |
||||
|
str, Depends(partial(callable_dependency, "partial-callable-dependency")) |
||||
|
], |
||||
|
) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/partial-callable-gen-dependency") |
||||
|
async def get_partial_callable_gen_dependency( |
||||
|
value: Annotated[ |
||||
|
str, |
||||
|
Depends(partial(callable_gen_dependency, "partial-callable-gen-dependency")), |
||||
|
], |
||||
|
) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/partial-async-callable-dependency") |
||||
|
async def get_partial_async_callable_dependency( |
||||
|
value: Annotated[ |
||||
|
str, |
||||
|
Depends( |
||||
|
partial(async_callable_dependency, "partial-async-callable-dependency") |
||||
|
), |
||||
|
], |
||||
|
) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/partial-async-callable-gen-dependency") |
||||
|
async def get_partial_async_callable_gen_dependency( |
||||
|
value: Annotated[ |
||||
|
str, |
||||
|
Depends( |
||||
|
partial( |
||||
|
async_callable_gen_dependency, "partial-async-callable-gen-dependency" |
||||
|
) |
||||
|
), |
||||
|
], |
||||
|
) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/partial-synchronous-method-dependency") |
||||
|
async def get_partial_synchronous_method_dependency( |
||||
|
value: Annotated[ |
||||
|
str, |
||||
|
Depends( |
||||
|
partial( |
||||
|
methods_dependency.synchronous, "partial-synchronous-method-dependency" |
||||
|
) |
||||
|
), |
||||
|
], |
||||
|
) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/partial-synchronous-method-gen-dependency") |
||||
|
async def get_partial_synchronous_method_gen_dependency( |
||||
|
value: Annotated[ |
||||
|
str, |
||||
|
Depends( |
||||
|
partial( |
||||
|
methods_dependency.synchronous_gen, |
||||
|
"partial-synchronous-method-gen-dependency", |
||||
|
) |
||||
|
), |
||||
|
], |
||||
|
) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/partial-asynchronous-method-dependency") |
||||
|
async def get_partial_asynchronous_method_dependency( |
||||
|
value: Annotated[ |
||||
|
str, |
||||
|
Depends( |
||||
|
partial( |
||||
|
methods_dependency.asynchronous, |
||||
|
"partial-asynchronous-method-dependency", |
||||
|
) |
||||
|
), |
||||
|
], |
||||
|
) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/partial-asynchronous-method-gen-dependency") |
||||
|
async def get_partial_asynchronous_method_gen_dependency( |
||||
|
value: Annotated[ |
||||
|
str, |
||||
|
Depends( |
||||
|
partial( |
||||
|
methods_dependency.asynchronous_gen, |
||||
|
"partial-asynchronous-method-gen-dependency", |
||||
|
) |
||||
|
), |
||||
|
], |
||||
|
) -> str: |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
@pytest.mark.parametrize( |
||||
|
"route,value", |
||||
|
[ |
||||
|
("/partial-function-dependency", "partial-function-dependency"), |
||||
|
( |
||||
|
"/partial-async-function-dependency", |
||||
|
"partial-async-function-dependency", |
||||
|
), |
||||
|
("/partial-gen-dependency", "partial-gen-dependency"), |
||||
|
("/partial-async-gen-dependency", "partial-async-gen-dependency"), |
||||
|
("/partial-callable-dependency", "partial-callable-dependency"), |
||||
|
("/partial-callable-gen-dependency", "partial-callable-gen-dependency"), |
||||
|
("/partial-async-callable-dependency", "partial-async-callable-dependency"), |
||||
|
( |
||||
|
"/partial-async-callable-gen-dependency", |
||||
|
"partial-async-callable-gen-dependency", |
||||
|
), |
||||
|
( |
||||
|
"/partial-synchronous-method-dependency", |
||||
|
"partial-synchronous-method-dependency", |
||||
|
), |
||||
|
( |
||||
|
"/partial-synchronous-method-gen-dependency", |
||||
|
"partial-synchronous-method-gen-dependency", |
||||
|
), |
||||
|
( |
||||
|
"/partial-asynchronous-method-dependency", |
||||
|
"partial-asynchronous-method-dependency", |
||||
|
), |
||||
|
( |
||||
|
"/partial-asynchronous-method-gen-dependency", |
||||
|
"partial-asynchronous-method-gen-dependency", |
||||
|
), |
||||
|
], |
||||
|
) |
||||
|
def test_dependency_types_with_partial(route: str, value: str) -> None: |
||||
|
response = client.get(route) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == value |
||||
@ -0,0 +1,449 @@ |
|||||
|
import inspect |
||||
|
import sys |
||||
|
from functools import wraps |
||||
|
from typing import AsyncGenerator, Generator |
||||
|
|
||||
|
import pytest |
||||
|
from fastapi import Depends, FastAPI |
||||
|
from fastapi.concurrency import iterate_in_threadpool, run_in_threadpool |
||||
|
from fastapi.testclient import TestClient |
||||
|
|
||||
|
if sys.version_info >= (3, 13): # pragma: no cover |
||||
|
from inspect import iscoroutinefunction |
||||
|
else: # pragma: no cover |
||||
|
from asyncio import iscoroutinefunction |
||||
|
|
||||
|
|
||||
|
def noop_wrap(func): |
||||
|
@wraps(func) |
||||
|
def wrapper(*args, **kwargs): |
||||
|
return func(*args, **kwargs) |
||||
|
|
||||
|
return wrapper |
||||
|
|
||||
|
|
||||
|
def noop_wrap_async(func): |
||||
|
if inspect.isgeneratorfunction(func): |
||||
|
|
||||
|
@wraps(func) |
||||
|
async def gen_wrapper(*args, **kwargs): |
||||
|
async for item in iterate_in_threadpool(func(*args, **kwargs)): |
||||
|
yield item |
||||
|
|
||||
|
return gen_wrapper |
||||
|
|
||||
|
elif inspect.isasyncgenfunction(func): |
||||
|
|
||||
|
@wraps(func) |
||||
|
async def async_gen_wrapper(*args, **kwargs): |
||||
|
async for item in func(*args, **kwargs): |
||||
|
yield item |
||||
|
|
||||
|
return async_gen_wrapper |
||||
|
|
||||
|
@wraps(func) |
||||
|
async def wrapper(*args, **kwargs): |
||||
|
if inspect.isroutine(func) and iscoroutinefunction(func): |
||||
|
return await func(*args, **kwargs) |
||||
|
if inspect.isclass(func): |
||||
|
return await run_in_threadpool(func, *args, **kwargs) |
||||
|
dunder_call = getattr(func, "__call__", None) # noqa: B004 |
||||
|
if iscoroutinefunction(dunder_call): |
||||
|
return await dunder_call(*args, **kwargs) |
||||
|
return await run_in_threadpool(func, *args, **kwargs) |
||||
|
|
||||
|
return wrapper |
||||
|
|
||||
|
|
||||
|
class ClassInstanceDep: |
||||
|
def __call__(self): |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
class_instance_dep = ClassInstanceDep() |
||||
|
wrapped_class_instance_dep = noop_wrap(class_instance_dep) |
||||
|
wrapped_class_instance_dep_async_wrapper = noop_wrap_async(class_instance_dep) |
||||
|
|
||||
|
|
||||
|
class ClassInstanceGenDep: |
||||
|
def __call__(self): |
||||
|
yield True |
||||
|
|
||||
|
|
||||
|
class_instance_gen_dep = ClassInstanceGenDep() |
||||
|
wrapped_class_instance_gen_dep = noop_wrap(class_instance_gen_dep) |
||||
|
|
||||
|
|
||||
|
class ClassInstanceWrappedDep: |
||||
|
@noop_wrap |
||||
|
def __call__(self): |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
class_instance_wrapped_dep = ClassInstanceWrappedDep() |
||||
|
|
||||
|
|
||||
|
class ClassInstanceWrappedAsyncDep: |
||||
|
@noop_wrap_async |
||||
|
def __call__(self): |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
class_instance_wrapped_async_dep = ClassInstanceWrappedAsyncDep() |
||||
|
|
||||
|
|
||||
|
class ClassInstanceWrappedGenDep: |
||||
|
@noop_wrap |
||||
|
def __call__(self): |
||||
|
yield True |
||||
|
|
||||
|
|
||||
|
class_instance_wrapped_gen_dep = ClassInstanceWrappedGenDep() |
||||
|
|
||||
|
|
||||
|
class ClassInstanceWrappedAsyncGenDep: |
||||
|
@noop_wrap_async |
||||
|
def __call__(self): |
||||
|
yield True |
||||
|
|
||||
|
|
||||
|
class_instance_wrapped_async_gen_dep = ClassInstanceWrappedAsyncGenDep() |
||||
|
|
||||
|
|
||||
|
class ClassDep: |
||||
|
def __init__(self): |
||||
|
self.value = True |
||||
|
|
||||
|
|
||||
|
wrapped_class_dep = noop_wrap(ClassDep) |
||||
|
wrapped_class_dep_async_wrapper = noop_wrap_async(ClassDep) |
||||
|
|
||||
|
|
||||
|
class ClassInstanceAsyncDep: |
||||
|
async def __call__(self): |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
class_instance_async_dep = ClassInstanceAsyncDep() |
||||
|
wrapped_class_instance_async_dep = noop_wrap(class_instance_async_dep) |
||||
|
wrapped_class_instance_async_dep_async_wrapper = noop_wrap_async( |
||||
|
class_instance_async_dep |
||||
|
) |
||||
|
|
||||
|
|
||||
|
class ClassInstanceAsyncGenDep: |
||||
|
async def __call__(self): |
||||
|
yield True |
||||
|
|
||||
|
|
||||
|
class_instance_async_gen_dep = ClassInstanceAsyncGenDep() |
||||
|
wrapped_class_instance_async_gen_dep = noop_wrap(class_instance_async_gen_dep) |
||||
|
|
||||
|
|
||||
|
class ClassInstanceAsyncWrappedDep: |
||||
|
@noop_wrap |
||||
|
async def __call__(self): |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
class_instance_async_wrapped_dep = ClassInstanceAsyncWrappedDep() |
||||
|
|
||||
|
|
||||
|
class ClassInstanceAsyncWrappedAsyncDep: |
||||
|
@noop_wrap_async |
||||
|
async def __call__(self): |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
class_instance_async_wrapped_async_dep = ClassInstanceAsyncWrappedAsyncDep() |
||||
|
|
||||
|
|
||||
|
class ClassInstanceAsyncWrappedGenDep: |
||||
|
@noop_wrap |
||||
|
async def __call__(self): |
||||
|
yield True |
||||
|
|
||||
|
|
||||
|
class_instance_async_wrapped_gen_dep = ClassInstanceAsyncWrappedGenDep() |
||||
|
|
||||
|
|
||||
|
class ClassInstanceAsyncWrappedGenAsyncDep: |
||||
|
@noop_wrap_async |
||||
|
async def __call__(self): |
||||
|
yield True |
||||
|
|
||||
|
|
||||
|
class_instance_async_wrapped_gen_async_dep = ClassInstanceAsyncWrappedGenAsyncDep() |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
# Sync wrapper |
||||
|
|
||||
|
|
||||
|
@noop_wrap |
||||
|
def wrapped_dependency() -> bool: |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
@noop_wrap |
||||
|
def wrapped_gen_dependency() -> Generator[bool, None, None]: |
||||
|
yield True |
||||
|
|
||||
|
|
||||
|
@noop_wrap |
||||
|
async def async_wrapped_dependency() -> bool: |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
@noop_wrap |
||||
|
async def async_wrapped_gen_dependency() -> AsyncGenerator[bool, None]: |
||||
|
yield True |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-dependency/") |
||||
|
async def get_wrapped_dependency(value: bool = Depends(wrapped_dependency)): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-gen-dependency/") |
||||
|
async def get_wrapped_gen_dependency(value: bool = Depends(wrapped_gen_dependency)): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/async-wrapped-dependency/") |
||||
|
async def get_async_wrapped_dependency(value: bool = Depends(async_wrapped_dependency)): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/async-wrapped-gen-dependency/") |
||||
|
async def get_async_wrapped_gen_dependency( |
||||
|
value: bool = Depends(async_wrapped_gen_dependency), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-class-instance-dependency/") |
||||
|
async def get_wrapped_class_instance_dependency( |
||||
|
value: bool = Depends(wrapped_class_instance_dep), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-class-instance-async-dependency/") |
||||
|
async def get_wrapped_class_instance_async_dependency( |
||||
|
value: bool = Depends(wrapped_class_instance_async_dep), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-class-instance-gen-dependency/") |
||||
|
async def get_wrapped_class_instance_gen_dependency( |
||||
|
value: bool = Depends(wrapped_class_instance_gen_dep), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-class-instance-async-gen-dependency/") |
||||
|
async def get_wrapped_class_instance_async_gen_dependency( |
||||
|
value: bool = Depends(wrapped_class_instance_async_gen_dep), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/class-instance-wrapped-dependency/") |
||||
|
async def get_class_instance_wrapped_dependency( |
||||
|
value: bool = Depends(class_instance_wrapped_dep), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/class-instance-wrapped-async-dependency/") |
||||
|
async def get_class_instance_wrapped_async_dependency( |
||||
|
value: bool = Depends(class_instance_wrapped_async_dep), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/class-instance-async-wrapped-dependency/") |
||||
|
async def get_class_instance_async_wrapped_dependency( |
||||
|
value: bool = Depends(class_instance_async_wrapped_dep), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/class-instance-async-wrapped-async-dependency/") |
||||
|
async def get_class_instance_async_wrapped_async_dependency( |
||||
|
value: bool = Depends(class_instance_async_wrapped_async_dep), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/class-instance-wrapped-gen-dependency/") |
||||
|
async def get_class_instance_wrapped_gen_dependency( |
||||
|
value: bool = Depends(class_instance_wrapped_gen_dep), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/class-instance-wrapped-async-gen-dependency/") |
||||
|
async def get_class_instance_wrapped_async_gen_dependency( |
||||
|
value: bool = Depends(class_instance_wrapped_async_gen_dep), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/class-instance-async-wrapped-gen-dependency/") |
||||
|
async def get_class_instance_async_wrapped_gen_dependency( |
||||
|
value: bool = Depends(class_instance_async_wrapped_gen_dep), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/class-instance-async-wrapped-gen-async-dependency/") |
||||
|
async def get_class_instance_async_wrapped_gen_async_dependency( |
||||
|
value: bool = Depends(class_instance_async_wrapped_gen_async_dep), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-class-dependency/") |
||||
|
async def get_wrapped_class_dependency(value: ClassDep = Depends(wrapped_class_dep)): |
||||
|
return value.value |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-endpoint/") |
||||
|
@noop_wrap |
||||
|
def get_wrapped_endpoint(): |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
@app.get("/async-wrapped-endpoint/") |
||||
|
@noop_wrap |
||||
|
async def get_async_wrapped_endpoint(): |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
# Async wrapper |
||||
|
|
||||
|
|
||||
|
@noop_wrap_async |
||||
|
def wrapped_dependency_async_wrapper() -> bool: |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
@noop_wrap_async |
||||
|
def wrapped_gen_dependency_async_wrapper() -> Generator[bool, None, None]: |
||||
|
yield True |
||||
|
|
||||
|
|
||||
|
@noop_wrap_async |
||||
|
async def async_wrapped_dependency_async_wrapper() -> bool: |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
@noop_wrap_async |
||||
|
async def async_wrapped_gen_dependency_async_wrapper() -> AsyncGenerator[bool, None]: |
||||
|
yield True |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-dependency-async-wrapper/") |
||||
|
async def get_wrapped_dependency_async_wrapper( |
||||
|
value: bool = Depends(wrapped_dependency_async_wrapper), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-gen-dependency-async-wrapper/") |
||||
|
async def get_wrapped_gen_dependency_async_wrapper( |
||||
|
value: bool = Depends(wrapped_gen_dependency_async_wrapper), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/async-wrapped-dependency-async-wrapper/") |
||||
|
async def get_async_wrapped_dependency_async_wrapper( |
||||
|
value: bool = Depends(async_wrapped_dependency_async_wrapper), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/async-wrapped-gen-dependency-async-wrapper/") |
||||
|
async def get_async_wrapped_gen_dependency_async_wrapper( |
||||
|
value: bool = Depends(async_wrapped_gen_dependency_async_wrapper), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-class-instance-dependency-async-wrapper/") |
||||
|
async def get_wrapped_class_instance_dependency_async_wrapper( |
||||
|
value: bool = Depends(wrapped_class_instance_dep_async_wrapper), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-class-instance-async-dependency-async-wrapper/") |
||||
|
async def get_wrapped_class_instance_async_dependency_async_wrapper( |
||||
|
value: bool = Depends(wrapped_class_instance_async_dep_async_wrapper), |
||||
|
): |
||||
|
return value |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-class-dependency-async-wrapper/") |
||||
|
async def get_wrapped_class_dependency_async_wrapper( |
||||
|
value: ClassDep = Depends(wrapped_class_dep_async_wrapper), |
||||
|
): |
||||
|
return value.value |
||||
|
|
||||
|
|
||||
|
@app.get("/wrapped-endpoint-async-wrapper/") |
||||
|
@noop_wrap_async |
||||
|
def get_wrapped_endpoint_async_wrapper(): |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
@app.get("/async-wrapped-endpoint-async-wrapper/") |
||||
|
@noop_wrap_async |
||||
|
async def get_async_wrapped_endpoint_async_wrapper(): |
||||
|
return True |
||||
|
|
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
@pytest.mark.parametrize( |
||||
|
"route", |
||||
|
[ |
||||
|
"/wrapped-dependency/", |
||||
|
"/wrapped-gen-dependency/", |
||||
|
"/async-wrapped-dependency/", |
||||
|
"/async-wrapped-gen-dependency/", |
||||
|
"/wrapped-class-instance-dependency/", |
||||
|
"/wrapped-class-instance-async-dependency/", |
||||
|
"/wrapped-class-instance-gen-dependency/", |
||||
|
"/wrapped-class-instance-async-gen-dependency/", |
||||
|
"/class-instance-wrapped-dependency/", |
||||
|
"/class-instance-wrapped-async-dependency/", |
||||
|
"/class-instance-async-wrapped-dependency/", |
||||
|
"/class-instance-async-wrapped-async-dependency/", |
||||
|
"/class-instance-wrapped-gen-dependency/", |
||||
|
"/class-instance-wrapped-async-gen-dependency/", |
||||
|
"/class-instance-async-wrapped-gen-dependency/", |
||||
|
"/class-instance-async-wrapped-gen-async-dependency/", |
||||
|
"/wrapped-class-dependency/", |
||||
|
"/wrapped-endpoint/", |
||||
|
"/async-wrapped-endpoint/", |
||||
|
"/wrapped-dependency-async-wrapper/", |
||||
|
"/wrapped-gen-dependency-async-wrapper/", |
||||
|
"/async-wrapped-dependency-async-wrapper/", |
||||
|
"/async-wrapped-gen-dependency-async-wrapper/", |
||||
|
"/wrapped-class-instance-dependency-async-wrapper/", |
||||
|
"/wrapped-class-instance-async-dependency-async-wrapper/", |
||||
|
"/wrapped-class-dependency-async-wrapper/", |
||||
|
"/wrapped-endpoint-async-wrapper/", |
||||
|
"/async-wrapped-endpoint-async-wrapper/", |
||||
|
], |
||||
|
) |
||||
|
def test_class_dependency(route): |
||||
|
response = client.get(route) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() is True |
||||
@ -0,0 +1,35 @@ |
|||||
|
from typing import Optional |
||||
|
|
||||
|
from fastapi import FastAPI, File, Form |
||||
|
from starlette.testclient import TestClient |
||||
|
from typing_extensions import Annotated |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.post("/urlencoded") |
||||
|
async def post_url_encoded(age: Annotated[Optional[int], Form()] = None): |
||||
|
return age |
||||
|
|
||||
|
|
||||
|
@app.post("/multipart") |
||||
|
async def post_multi_part( |
||||
|
age: Annotated[Optional[int], Form()] = None, |
||||
|
file: Annotated[Optional[bytes], File()] = None, |
||||
|
): |
||||
|
return {"file": file, "age": age} |
||||
|
|
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_form_default_url_encoded(): |
||||
|
response = client.post("/urlencoded", data={"age": ""}) |
||||
|
assert response.status_code == 200 |
||||
|
assert response.text == "null" |
||||
|
|
||||
|
|
||||
|
def test_form_default_multi_part(): |
||||
|
response = client.post("/multipart", data={"age": ""}) |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"file": None, "age": None} |
||||
@ -0,0 +1,30 @@ |
|||||
|
from typing import List, Optional |
||||
|
|
||||
|
from fastapi import FastAPI, File |
||||
|
from fastapi.testclient import TestClient |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.post("/files") |
||||
|
async def upload_files(files: Optional[List[bytes]] = File(None)): |
||||
|
if files is None: |
||||
|
return {"files_count": 0} |
||||
|
return {"files_count": len(files), "sizes": [len(f) for f in files]} |
||||
|
|
||||
|
|
||||
|
def test_optional_bytes_list(): |
||||
|
client = TestClient(app) |
||||
|
response = client.post( |
||||
|
"/files", |
||||
|
files=[("files", b"content1"), ("files", b"content2")], |
||||
|
) |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"files_count": 2, "sizes": [8, 8]} |
||||
|
|
||||
|
|
||||
|
def test_optional_bytes_list_no_files(): |
||||
|
client = TestClient(app) |
||||
|
response = client.post("/files") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"files_count": 0} |
||||
@ -0,0 +1,111 @@ |
|||||
|
from fastapi import Cookie, FastAPI, Header, Query |
||||
|
from fastapi._compat import PYDANTIC_V2 |
||||
|
from fastapi.testclient import TestClient |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class Model(BaseModel): |
||||
|
param: str |
||||
|
|
||||
|
if PYDANTIC_V2: |
||||
|
model_config = {"extra": "allow"} |
||||
|
else: |
||||
|
|
||||
|
class Config: |
||||
|
extra = "allow" |
||||
|
|
||||
|
|
||||
|
@app.get("/query") |
||||
|
async def query_model_with_extra(data: Model = Query()): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
@app.get("/header") |
||||
|
async def header_model_with_extra(data: Model = Header()): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
@app.get("/cookie") |
||||
|
async def cookies_model_with_extra(data: Model = Cookie()): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
def test_query_pass_extra_list(): |
||||
|
client = TestClient(app) |
||||
|
resp = client.get( |
||||
|
"/query", |
||||
|
params={ |
||||
|
"param": "123", |
||||
|
"param2": ["456", "789"], # Pass a list of values as extra parameter |
||||
|
}, |
||||
|
) |
||||
|
assert resp.status_code == 200 |
||||
|
assert resp.json() == { |
||||
|
"param": "123", |
||||
|
"param2": ["456", "789"], |
||||
|
} |
||||
|
|
||||
|
|
||||
|
def test_query_pass_extra_single(): |
||||
|
client = TestClient(app) |
||||
|
resp = client.get( |
||||
|
"/query", |
||||
|
params={ |
||||
|
"param": "123", |
||||
|
"param2": "456", |
||||
|
}, |
||||
|
) |
||||
|
assert resp.status_code == 200 |
||||
|
assert resp.json() == { |
||||
|
"param": "123", |
||||
|
"param2": "456", |
||||
|
} |
||||
|
|
||||
|
|
||||
|
def test_header_pass_extra_list(): |
||||
|
client = TestClient(app) |
||||
|
|
||||
|
resp = client.get( |
||||
|
"/header", |
||||
|
headers=[ |
||||
|
("param", "123"), |
||||
|
("param2", "456"), # Pass a list of values as extra parameter |
||||
|
("param2", "789"), |
||||
|
], |
||||
|
) |
||||
|
assert resp.status_code == 200 |
||||
|
resp_json = resp.json() |
||||
|
assert "param2" in resp_json |
||||
|
assert resp_json["param2"] == ["456", "789"] |
||||
|
|
||||
|
|
||||
|
def test_header_pass_extra_single(): |
||||
|
client = TestClient(app) |
||||
|
|
||||
|
resp = client.get( |
||||
|
"/header", |
||||
|
headers=[ |
||||
|
("param", "123"), |
||||
|
("param2", "456"), |
||||
|
], |
||||
|
) |
||||
|
assert resp.status_code == 200 |
||||
|
resp_json = resp.json() |
||||
|
assert "param2" in resp_json |
||||
|
assert resp_json["param2"] == "456" |
||||
|
|
||||
|
|
||||
|
def test_cookie_pass_extra_list(): |
||||
|
client = TestClient(app) |
||||
|
client.cookies = [ |
||||
|
("param", "123"), |
||||
|
("param2", "456"), # Pass a list of values as extra parameter |
||||
|
("param2", "789"), |
||||
|
] |
||||
|
resp = client.get("/cookie") |
||||
|
assert resp.status_code == 200 |
||||
|
resp_json = resp.json() |
||||
|
assert "param2" in resp_json |
||||
|
assert resp_json["param2"] == "789" # Cookies only keep the last value |
||||
@ -0,0 +1,76 @@ |
|||||
|
from dirty_equals import IsPartialDict |
||||
|
from fastapi import Cookie, FastAPI, Header, Query |
||||
|
from fastapi._compat import PYDANTIC_V2 |
||||
|
from fastapi.testclient import TestClient |
||||
|
from pydantic import BaseModel, Field |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class Model(BaseModel): |
||||
|
param: str = Field(alias="param_alias") |
||||
|
|
||||
|
|
||||
|
@app.get("/query") |
||||
|
async def query_model(data: Model = Query()): |
||||
|
return {"param": data.param} |
||||
|
|
||||
|
|
||||
|
@app.get("/header") |
||||
|
async def header_model(data: Model = Header()): |
||||
|
return {"param": data.param} |
||||
|
|
||||
|
|
||||
|
@app.get("/cookie") |
||||
|
async def cookie_model(data: Model = Cookie()): |
||||
|
return {"param": data.param} |
||||
|
|
||||
|
|
||||
|
def test_query_model_with_alias(): |
||||
|
client = TestClient(app) |
||||
|
response = client.get("/query", params={"param_alias": "value"}) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"param": "value"} |
||||
|
|
||||
|
|
||||
|
def test_header_model_with_alias(): |
||||
|
client = TestClient(app) |
||||
|
response = client.get("/header", headers={"param_alias": "value"}) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"param": "value"} |
||||
|
|
||||
|
|
||||
|
def test_cookie_model_with_alias(): |
||||
|
client = TestClient(app) |
||||
|
client.cookies.set("param_alias", "value") |
||||
|
response = client.get("/cookie") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"param": "value"} |
||||
|
|
||||
|
|
||||
|
def test_query_model_with_alias_by_name(): |
||||
|
client = TestClient(app) |
||||
|
response = client.get("/query", params={"param": "value"}) |
||||
|
assert response.status_code == 422, response.text |
||||
|
details = response.json() |
||||
|
if PYDANTIC_V2: |
||||
|
assert details["detail"][0]["input"] == {"param": "value"} |
||||
|
|
||||
|
|
||||
|
def test_header_model_with_alias_by_name(): |
||||
|
client = TestClient(app) |
||||
|
response = client.get("/header", headers={"param": "value"}) |
||||
|
assert response.status_code == 422, response.text |
||||
|
details = response.json() |
||||
|
if PYDANTIC_V2: |
||||
|
assert details["detail"][0]["input"] == IsPartialDict({"param": "value"}) |
||||
|
|
||||
|
|
||||
|
def test_cookie_model_with_alias_by_name(): |
||||
|
client = TestClient(app) |
||||
|
client.cookies.set("param", "value") |
||||
|
response = client.get("/cookie") |
||||
|
assert response.status_code == 422, response.text |
||||
|
details = response.json() |
||||
|
if PYDANTIC_V2: |
||||
|
assert details["detail"][0]["input"] == {"param": "value"} |
||||
@ -0,0 +1,92 @@ |
|||||
|
import pytest |
||||
|
from fastapi import FastAPI |
||||
|
from fastapi.testclient import TestClient |
||||
|
from inline_snapshot import snapshot |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
from tests.utils import needs_py310, needs_pydanticv2 |
||||
|
|
||||
|
|
||||
|
@pytest.fixture(name="client") |
||||
|
def get_client(): |
||||
|
from enum import Enum |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
class PlatformRole(str, Enum): |
||||
|
admin = "admin" |
||||
|
user = "user" |
||||
|
|
||||
|
class OtherRole(str, Enum): ... |
||||
|
|
||||
|
class User(BaseModel): |
||||
|
username: str |
||||
|
role: PlatformRole | OtherRole |
||||
|
|
||||
|
@app.get("/users") |
||||
|
async def get_user() -> User: |
||||
|
return {"username": "alice", "role": "admin"} |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
return client |
||||
|
|
||||
|
|
||||
|
@needs_py310 |
||||
|
@needs_pydanticv2 |
||||
|
def test_get(client: TestClient): |
||||
|
response = client.get("/users") |
||||
|
assert response.json() == {"username": "alice", "role": "admin"} |
||||
|
|
||||
|
|
||||
|
@needs_py310 |
||||
|
@needs_pydanticv2 |
||||
|
def test_openapi_schema(client: TestClient): |
||||
|
response = client.get("openapi.json") |
||||
|
assert response.json() == snapshot( |
||||
|
{ |
||||
|
"openapi": "3.1.0", |
||||
|
"info": {"title": "FastAPI", "version": "0.1.0"}, |
||||
|
"paths": { |
||||
|
"/users": { |
||||
|
"get": { |
||||
|
"summary": "Get User", |
||||
|
"operationId": "get_user_users_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": { |
||||
|
"application/json": { |
||||
|
"schema": {"$ref": "#/components/schemas/User"} |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
"components": { |
||||
|
"schemas": { |
||||
|
"PlatformRole": { |
||||
|
"type": "string", |
||||
|
"enum": ["admin", "user"], |
||||
|
"title": "PlatformRole", |
||||
|
}, |
||||
|
"User": { |
||||
|
"properties": { |
||||
|
"username": {"type": "string", "title": "Username"}, |
||||
|
"role": { |
||||
|
"anyOf": [ |
||||
|
{"$ref": "#/components/schemas/PlatformRole"}, |
||||
|
{"enum": [], "title": "OtherRole"}, |
||||
|
], |
||||
|
"title": "Role", |
||||
|
}, |
||||
|
}, |
||||
|
"type": "object", |
||||
|
"required": ["username", "role"], |
||||
|
"title": "User", |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
) |
||||
@ -0,0 +1,198 @@ |
|||||
|
# Ref: https://github.com/fastapi/fastapi/issues/14454 |
||||
|
|
||||
|
from typing import Optional |
||||
|
|
||||
|
from fastapi import APIRouter, Depends, FastAPI, Security |
||||
|
from fastapi.security import OAuth2AuthorizationCodeBearer |
||||
|
from fastapi.testclient import TestClient |
||||
|
from inline_snapshot import snapshot |
||||
|
from typing_extensions import Annotated |
||||
|
|
||||
|
oauth2_scheme = OAuth2AuthorizationCodeBearer( |
||||
|
authorizationUrl="authorize", |
||||
|
tokenUrl="token", |
||||
|
auto_error=True, |
||||
|
scopes={"read": "Read access", "write": "Write access"}, |
||||
|
) |
||||
|
|
||||
|
|
||||
|
async def get_token(token: Annotated[str, Depends(oauth2_scheme)]) -> str: |
||||
|
return token |
||||
|
|
||||
|
|
||||
|
app = FastAPI(dependencies=[Depends(get_token)]) |
||||
|
|
||||
|
|
||||
|
@app.get("/") |
||||
|
async def root(): |
||||
|
return {"message": "Hello World"} |
||||
|
|
||||
|
|
||||
|
@app.get( |
||||
|
"/with-oauth2-scheme", |
||||
|
dependencies=[Security(oauth2_scheme, scopes=["read", "write"])], |
||||
|
) |
||||
|
async def read_with_oauth2_scheme(): |
||||
|
return {"message": "Admin Access"} |
||||
|
|
||||
|
|
||||
|
@app.get( |
||||
|
"/with-get-token", dependencies=[Security(get_token, scopes=["read", "write"])] |
||||
|
) |
||||
|
async def read_with_get_token(): |
||||
|
return {"message": "Admin Access"} |
||||
|
|
||||
|
|
||||
|
router = APIRouter(dependencies=[Security(oauth2_scheme, scopes=["read"])]) |
||||
|
|
||||
|
|
||||
|
@router.get("/items/") |
||||
|
async def read_items(token: Optional[str] = Depends(oauth2_scheme)): |
||||
|
return {"token": token} |
||||
|
|
||||
|
|
||||
|
@router.post("/items/") |
||||
|
async def create_item( |
||||
|
token: Optional[str] = Security(oauth2_scheme, scopes=["read", "write"]), |
||||
|
): |
||||
|
return {"token": token} |
||||
|
|
||||
|
|
||||
|
app.include_router(router) |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_root(): |
||||
|
response = client.get("/", headers={"Authorization": "Bearer testtoken"}) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"message": "Hello World"} |
||||
|
|
||||
|
|
||||
|
def test_read_with_oauth2_scheme(): |
||||
|
response = client.get( |
||||
|
"/with-oauth2-scheme", headers={"Authorization": "Bearer testtoken"} |
||||
|
) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"message": "Admin Access"} |
||||
|
|
||||
|
|
||||
|
def test_read_with_get_token(): |
||||
|
response = client.get( |
||||
|
"/with-get-token", headers={"Authorization": "Bearer testtoken"} |
||||
|
) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"message": "Admin Access"} |
||||
|
|
||||
|
|
||||
|
def test_read_token(): |
||||
|
response = client.get("/items/", headers={"Authorization": "Bearer testtoken"}) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"token": "testtoken"} |
||||
|
|
||||
|
|
||||
|
def test_create_token(): |
||||
|
response = client.post("/items/", headers={"Authorization": "Bearer testtoken"}) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"token": "testtoken"} |
||||
|
|
||||
|
|
||||
|
def test_openapi_schema(): |
||||
|
response = client.get("/openapi.json") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == snapshot( |
||||
|
{ |
||||
|
"openapi": "3.1.0", |
||||
|
"info": {"title": "FastAPI", "version": "0.1.0"}, |
||||
|
"paths": { |
||||
|
"/": { |
||||
|
"get": { |
||||
|
"summary": "Root", |
||||
|
"operationId": "root__get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": {"application/json": {"schema": {}}}, |
||||
|
} |
||||
|
}, |
||||
|
"security": [{"OAuth2AuthorizationCodeBearer": []}], |
||||
|
} |
||||
|
}, |
||||
|
"/with-oauth2-scheme": { |
||||
|
"get": { |
||||
|
"summary": "Read With Oauth2 Scheme", |
||||
|
"operationId": "read_with_oauth2_scheme_with_oauth2_scheme_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": {"application/json": {"schema": {}}}, |
||||
|
} |
||||
|
}, |
||||
|
"security": [ |
||||
|
{"OAuth2AuthorizationCodeBearer": ["read", "write"]} |
||||
|
], |
||||
|
} |
||||
|
}, |
||||
|
"/with-get-token": { |
||||
|
"get": { |
||||
|
"summary": "Read With Get Token", |
||||
|
"operationId": "read_with_get_token_with_get_token_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": {"application/json": {"schema": {}}}, |
||||
|
} |
||||
|
}, |
||||
|
"security": [ |
||||
|
{"OAuth2AuthorizationCodeBearer": ["read", "write"]} |
||||
|
], |
||||
|
} |
||||
|
}, |
||||
|
"/items/": { |
||||
|
"get": { |
||||
|
"summary": "Read Items", |
||||
|
"operationId": "read_items_items__get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": {"application/json": {"schema": {}}}, |
||||
|
} |
||||
|
}, |
||||
|
"security": [ |
||||
|
{"OAuth2AuthorizationCodeBearer": ["read"]}, |
||||
|
], |
||||
|
}, |
||||
|
"post": { |
||||
|
"summary": "Create Item", |
||||
|
"operationId": "create_item_items__post", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": {"application/json": {"schema": {}}}, |
||||
|
} |
||||
|
}, |
||||
|
"security": [ |
||||
|
{"OAuth2AuthorizationCodeBearer": ["read", "write"]}, |
||||
|
], |
||||
|
}, |
||||
|
}, |
||||
|
}, |
||||
|
"components": { |
||||
|
"securitySchemes": { |
||||
|
"OAuth2AuthorizationCodeBearer": { |
||||
|
"type": "oauth2", |
||||
|
"flows": { |
||||
|
"authorizationCode": { |
||||
|
"scopes": { |
||||
|
"read": "Read access", |
||||
|
"write": "Write access", |
||||
|
}, |
||||
|
"authorizationUrl": "authorize", |
||||
|
"tokenUrl": "token", |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
) |
||||
@ -0,0 +1,79 @@ |
|||||
|
# Ref: https://github.com/fastapi/fastapi/issues/14454 |
||||
|
|
||||
|
from fastapi import Depends, FastAPI, Security |
||||
|
from fastapi.security import OAuth2AuthorizationCodeBearer |
||||
|
from fastapi.testclient import TestClient |
||||
|
from inline_snapshot import snapshot |
||||
|
from typing_extensions import Annotated |
||||
|
|
||||
|
oauth2_scheme = OAuth2AuthorizationCodeBearer( |
||||
|
authorizationUrl="api/oauth/authorize", |
||||
|
tokenUrl="/api/oauth/token", |
||||
|
scopes={"read": "Read access", "write": "Write access"}, |
||||
|
) |
||||
|
|
||||
|
|
||||
|
async def get_token(token: Annotated[str, Depends(oauth2_scheme)]) -> str: |
||||
|
return token |
||||
|
|
||||
|
|
||||
|
app = FastAPI(dependencies=[Depends(get_token)]) |
||||
|
|
||||
|
|
||||
|
@app.get("/admin", dependencies=[Security(get_token, scopes=["read", "write"])]) |
||||
|
async def read_admin(): |
||||
|
return {"message": "Admin Access"} |
||||
|
|
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_read_admin(): |
||||
|
response = client.get("/admin", headers={"Authorization": "Bearer faketoken"}) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"message": "Admin Access"} |
||||
|
|
||||
|
|
||||
|
def test_openapi_schema(): |
||||
|
response = client.get("/openapi.json") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == snapshot( |
||||
|
{ |
||||
|
"openapi": "3.1.0", |
||||
|
"info": {"title": "FastAPI", "version": "0.1.0"}, |
||||
|
"paths": { |
||||
|
"/admin": { |
||||
|
"get": { |
||||
|
"summary": "Read Admin", |
||||
|
"operationId": "read_admin_admin_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": {"application/json": {"schema": {}}}, |
||||
|
} |
||||
|
}, |
||||
|
"security": [ |
||||
|
{"OAuth2AuthorizationCodeBearer": ["read", "write"]} |
||||
|
], |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
"components": { |
||||
|
"securitySchemes": { |
||||
|
"OAuth2AuthorizationCodeBearer": { |
||||
|
"type": "oauth2", |
||||
|
"flows": { |
||||
|
"authorizationCode": { |
||||
|
"scopes": { |
||||
|
"read": "Read access", |
||||
|
"write": "Write access", |
||||
|
}, |
||||
|
"authorizationUrl": "api/oauth/authorize", |
||||
|
"tokenUrl": "/api/oauth/token", |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
) |
||||
@ -0,0 +1,46 @@ |
|||||
|
from typing import Dict |
||||
|
|
||||
|
import pytest |
||||
|
from fastapi import Depends, FastAPI, Security |
||||
|
from fastapi.testclient import TestClient |
||||
|
from typing_extensions import Annotated |
||||
|
|
||||
|
|
||||
|
@pytest.fixture(name="call_counter") |
||||
|
def call_counter_fixture(): |
||||
|
return {"count": 0} |
||||
|
|
||||
|
|
||||
|
@pytest.fixture(name="app") |
||||
|
def app_fixture(call_counter: Dict[str, int]): |
||||
|
def get_db(): |
||||
|
call_counter["count"] += 1 |
||||
|
return f"db_{call_counter['count']}" |
||||
|
|
||||
|
def get_user(db: Annotated[str, Depends(get_db)]): |
||||
|
return "user" |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
@app.get("/") |
||||
|
def endpoint( |
||||
|
db: Annotated[str, Depends(get_db)], |
||||
|
user: Annotated[str, Security(get_user, scopes=["read"])], |
||||
|
): |
||||
|
return {"db": db} |
||||
|
|
||||
|
return app |
||||
|
|
||||
|
|
||||
|
@pytest.fixture(name="client") |
||||
|
def client_fixture(app: FastAPI): |
||||
|
return TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_security_scopes_dependency_called_once( |
||||
|
client: TestClient, call_counter: Dict[str, int] |
||||
|
): |
||||
|
response = client.get("/") |
||||
|
|
||||
|
assert response.status_code == 200 |
||||
|
assert call_counter["count"] == 1 |
||||
@ -0,0 +1,45 @@ |
|||||
|
# Ref: https://github.com/tiangolo/fastapi/issues/5623 |
||||
|
|
||||
|
from typing import Any, Dict, List |
||||
|
|
||||
|
from fastapi import FastAPI, Security |
||||
|
from fastapi.security import SecurityScopes |
||||
|
from fastapi.testclient import TestClient |
||||
|
from typing_extensions import Annotated |
||||
|
|
||||
|
|
||||
|
async def security1(scopes: SecurityScopes): |
||||
|
return scopes.scopes |
||||
|
|
||||
|
|
||||
|
async def security2(scopes: SecurityScopes): |
||||
|
return scopes.scopes |
||||
|
|
||||
|
|
||||
|
async def dep3( |
||||
|
dep1: Annotated[List[str], Security(security1, scopes=["scope1"])], |
||||
|
dep2: Annotated[List[str], Security(security2, scopes=["scope2"])], |
||||
|
): |
||||
|
return {"dep1": dep1, "dep2": dep2} |
||||
|
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get("/scopes") |
||||
|
def get_scopes( |
||||
|
dep3: Annotated[Dict[str, Any], Security(dep3, scopes=["scope3"])], |
||||
|
): |
||||
|
return dep3 |
||||
|
|
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_security_scopes_dont_propagate(): |
||||
|
response = client.get("/scopes") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == { |
||||
|
"dep1": ["scope3", "scope1"], |
||||
|
"dep2": ["scope3", "scope2"], |
||||
|
} |
||||
@ -0,0 +1,107 @@ |
|||||
|
# Ref: https://github.com/fastapi/fastapi/discussions/6024#discussioncomment-8541913 |
||||
|
|
||||
|
from typing import Dict |
||||
|
|
||||
|
import pytest |
||||
|
from fastapi import Depends, FastAPI, Security |
||||
|
from fastapi.security import SecurityScopes |
||||
|
from fastapi.testclient import TestClient |
||||
|
from typing_extensions import Annotated |
||||
|
|
||||
|
|
||||
|
@pytest.fixture(name="call_counts") |
||||
|
def call_counts_fixture(): |
||||
|
return { |
||||
|
"get_db_session": 0, |
||||
|
"get_current_user": 0, |
||||
|
"get_user_me": 0, |
||||
|
"get_user_items": 0, |
||||
|
} |
||||
|
|
||||
|
|
||||
|
@pytest.fixture(name="app") |
||||
|
def app_fixture(call_counts: Dict[str, int]): |
||||
|
def get_db_session(): |
||||
|
call_counts["get_db_session"] += 1 |
||||
|
return f"db_session_{call_counts['get_db_session']}" |
||||
|
|
||||
|
def get_current_user( |
||||
|
security_scopes: SecurityScopes, |
||||
|
db_session: Annotated[str, Depends(get_db_session)], |
||||
|
): |
||||
|
call_counts["get_current_user"] += 1 |
||||
|
return { |
||||
|
"user": f"user_{call_counts['get_current_user']}", |
||||
|
"scopes": security_scopes.scopes, |
||||
|
"db_session": db_session, |
||||
|
} |
||||
|
|
||||
|
def get_user_me( |
||||
|
current_user: Annotated[dict, Security(get_current_user, scopes=["me"])], |
||||
|
): |
||||
|
call_counts["get_user_me"] += 1 |
||||
|
return { |
||||
|
"user_me": f"user_me_{call_counts['get_user_me']}", |
||||
|
"current_user": current_user, |
||||
|
} |
||||
|
|
||||
|
def get_user_items( |
||||
|
user_me: Annotated[dict, Depends(get_user_me)], |
||||
|
): |
||||
|
call_counts["get_user_items"] += 1 |
||||
|
return { |
||||
|
"user_items": f"user_items_{call_counts['get_user_items']}", |
||||
|
"user_me": user_me, |
||||
|
} |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
@app.get("/") |
||||
|
def path_operation( |
||||
|
user_me: Annotated[dict, Depends(get_user_me)], |
||||
|
user_items: Annotated[dict, Security(get_user_items, scopes=["items"])], |
||||
|
): |
||||
|
return { |
||||
|
"user_me": user_me, |
||||
|
"user_items": user_items, |
||||
|
} |
||||
|
|
||||
|
return app |
||||
|
|
||||
|
|
||||
|
@pytest.fixture(name="client") |
||||
|
def client_fixture(app: FastAPI): |
||||
|
return TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_security_scopes_sub_dependency_caching( |
||||
|
client: TestClient, call_counts: Dict[str, int] |
||||
|
): |
||||
|
response = client.get("/") |
||||
|
|
||||
|
assert response.status_code == 200 |
||||
|
assert call_counts["get_db_session"] == 1 |
||||
|
assert call_counts["get_current_user"] == 2 |
||||
|
assert call_counts["get_user_me"] == 2 |
||||
|
assert call_counts["get_user_items"] == 1 |
||||
|
assert response.json() == { |
||||
|
"user_me": { |
||||
|
"user_me": "user_me_1", |
||||
|
"current_user": { |
||||
|
"user": "user_1", |
||||
|
"scopes": ["me"], |
||||
|
"db_session": "db_session_1", |
||||
|
}, |
||||
|
}, |
||||
|
"user_items": { |
||||
|
"user_items": "user_items_1", |
||||
|
"user_me": { |
||||
|
"user_me": "user_me_2", |
||||
|
"current_user": { |
||||
|
"user": "user_2", |
||||
|
"scopes": ["items", "me"], |
||||
|
"db_session": "db_session_1", |
||||
|
}, |
||||
|
}, |
||||
|
}, |
||||
|
} |
||||
@ -0,0 +1,26 @@ |
|||||
|
from __future__ import annotations |
||||
|
|
||||
|
from fastapi import Depends, FastAPI, Request |
||||
|
from fastapi.testclient import TestClient |
||||
|
from typing_extensions import Annotated |
||||
|
|
||||
|
from .utils import needs_py310 |
||||
|
|
||||
|
|
||||
|
class Dep: |
||||
|
def __call__(self, request: Request): |
||||
|
return "test" |
||||
|
|
||||
|
|
||||
|
@needs_py310 |
||||
|
def test_stringified_annotations(): |
||||
|
app = FastAPI() |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
@app.get("/test/") |
||||
|
def call(test: Annotated[str, Depends(Dep())]): |
||||
|
return {"test": test} |
||||
|
|
||||
|
response = client.get("/test") |
||||
|
assert response.status_code == 200 |
||||
@ -0,0 +1,69 @@ |
|||||
|
import importlib |
||||
|
|
||||
|
import pytest |
||||
|
from fastapi.testclient import TestClient |
||||
|
from inline_snapshot import snapshot |
||||
|
|
||||
|
from ...utils import needs_py39 |
||||
|
|
||||
|
|
||||
|
@pytest.fixture( |
||||
|
name="client", |
||||
|
params=[ |
||||
|
"tutorial001_an", |
||||
|
pytest.param("tutorial001_an_py39", marks=needs_py39), |
||||
|
], |
||||
|
) |
||||
|
def get_client(request: pytest.FixtureRequest): |
||||
|
mod = importlib.import_module( |
||||
|
f"docs_src.authentication_error_status_code.{request.param}" |
||||
|
) |
||||
|
|
||||
|
client = TestClient(mod.app) |
||||
|
return client |
||||
|
|
||||
|
|
||||
|
def test_get_me(client: TestClient): |
||||
|
response = client.get("/me", headers={"Authorization": "Bearer secrettoken"}) |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == { |
||||
|
"message": "You are authenticated", |
||||
|
"token": "secrettoken", |
||||
|
} |
||||
|
|
||||
|
|
||||
|
def test_get_me_no_credentials(client: TestClient): |
||||
|
response = client.get("/me") |
||||
|
assert response.status_code == 403 |
||||
|
assert response.json() == {"detail": "Not authenticated"} |
||||
|
|
||||
|
|
||||
|
def test_openapi_schema(client: TestClient): |
||||
|
response = client.get("/openapi.json") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == snapshot( |
||||
|
{ |
||||
|
"openapi": "3.1.0", |
||||
|
"info": {"title": "FastAPI", "version": "0.1.0"}, |
||||
|
"paths": { |
||||
|
"/me": { |
||||
|
"get": { |
||||
|
"summary": "Read Me", |
||||
|
"operationId": "read_me_me_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": {"application/json": {"schema": {}}}, |
||||
|
} |
||||
|
}, |
||||
|
"security": [{"HTTPBearer403": []}], |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
"components": { |
||||
|
"securitySchemes": { |
||||
|
"HTTPBearer403": {"type": "http", "scheme": "bearer"} |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
) |
||||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue