--- /dev/null
+from fastapi import APIRouter, FastAPI
+from pydantic import BaseModel, HttpUrl
+from starlette.responses import JSONResponse
+
+app = FastAPI()
+
+
+class Invoice(BaseModel):
+ id: str
+ title: str = None
+ customer: str
+ total: float
+
+
+class InvoiceEvent(BaseModel):
+ description: str
+ paid: bool
+
+
+class InvoiceEventReceived(BaseModel):
+ ok: bool
+
+
+invoices_callback_router = APIRouter(default_response_class=JSONResponse)
+
+
+@invoices_callback_router.post(
+ "{$callback_url}/invoices/{$request.body.id}",
+ response_model=InvoiceEventReceived,
+)
+def invoice_notification(body: InvoiceEvent):
+ pass
+
+
+@app.post("/invoices/", callbacks=invoices_callback_router.routes)
+def create_invoice(invoice: Invoice, callback_url: HttpUrl = None):
+ """
+ Create an invoice.
+
+ This will (let's imagine) let the API user (some external developer) create an
+ invoice.
+
+ And this path operation will:
+
+ * Send the invoice to the client.
+ * Collect the money from the client.
+ * Send a notification back to the API user (the external developer), as a callback.
+ * At this point is that the API will somehow send a POST request to the
+ external API with the notification of the invoice event
+ (e.g. "payment successful").
+ """
+ # Send the invoice, collect the money, send the notification (the callback)
+ return {"msg": "Invoice received"}
--- /dev/null
+You could create an API with a *path operation* that could trigger a request to an *external API* created by someone else (probably the same developer that would be *using* your API).
+
+The process that happens when your API app calls the *external API* is named a "callback". Because the software that the external developer wrote sends a request to your API and then your API *calls back*, sending a request to an *external API* (that was probably created by the same developer).
+
+In this case, you could want to document how that external API *should* look like. What *path operation* it should have, what body it should expect, what response it should return, etc.
+
+## An app with callbacks
+
+Let's see all this with an example.
+
+Imagine you develop an app that allows creating invoices.
+
+These invoices will have an `id`, `title` (optional), `customer`, and `total`.
+
+The user of your API (an external developer) will create an invoice in your API with a POST request.
+
+Then your API will (let's imagine):
+
+* Send the invoice to some customer of the external developer.
+* Collect the money.
+* Send a notification back to the API user (the external developer).
+ * This will be done by sending a POST request (from *your API*) to some *external API* provided by that external developer (this is the "callback").
+
+## The normal **FastAPI** app
+
+Let's first see how the normal API app would look like before adding the callback.
+
+It will have a *path operation* that will receive an `Invoice` body, and a query parameter `callback_url` that will contain the URL for the callback.
+
+This part is pretty normal, most of the code is probably already familiar to you:
+
+```Python hl_lines="8 9 10 11 12 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54"
+{!./src/openapi_callbacks/tutorial001.py!}
+```
+
+!!! tip
+ The `callback_url` query parameter uses a Pydantic <a href="https://pydantic-docs.helpmanual.io/usage/types/#urls" target="_blank">URL</a> type.
+
+The only new thing is the `callbacks=messages_callback_router.routes` as an argument to the *path operation decorator*. We'll see what that is next.
+
+## Documenting the callback
+
+The actual callback code will depend heavily on your own API app.
+
+And it will probably vary a lot from one app to the next.
+
+It could be just one or two lines of code, like:
+
+```Python
+callback_url = "https://example.com/api/v1/invoices/events/"
+requests.post(callback_url, json={"description": "Invoice paid", "paid": True})
+```
+
+But possibly the most important part of the callback is making sure that your API user (the external developer) implements the *external API* correctly, according to the data that *your API* is going to send in the request body of the callback, etc.
+
+So, what we will do next is add the code to document how that *external API* should look like to receive the callback from *your API*.
+
+That documentation will show up in the Swagger UI at `/docs` in your API, and it will let external developers know how to build the *external API*.
+
+This example doesn't implement the callback itself (that could be just a line of code), only the documentation part.
+
+!!! tip
+ The actual callback is just an HTTP request.
+
+ When implementing the callback yourself, you could use something like <a href="https://www.encode.io/httpx/" target="_blank">HTTPX</a> or <a href="https://requests.readthedocs.io/" target="_blank">Requests</a>.
+
+## Write the callback documentation code
+
+This code won't be executed in your app, we only need it to *document* how that *external API* should look like.
+
+But, you already know how to easily create automatic documentation for an API with **FastAPI**.
+
+So we are going to use that same knowledge to document how the *external API* should look like... by creating the *path operation(s)* that the external API should implement (the ones your API will call).
+
+!!! tip
+ When writing the code to document a callback, it might be useful to imagine that you are that *external developer*. And that you are currently implementing the *external API*, not *your API*.
+
+ Temporarily adopting this point of view (of the *external developer*) can help you feel like it's more obvious where to put the parameters, the Pydantic model for the body, for the response, etc. for that *external API*.
+
+### Create a callback `APIRouter`
+
+First create a new `APIRouter` that will contain one or more callbacks.
+
+This router will never be added to an actual `FastAPI` app (i.e. it will never be passed to `app.include_router(...)`).
+
+Because of that, you need to declare what will be the `default_response_class`, and set it to `JSONResponse`.
+
+!!! Note "Technical Details"
+ The `response_class` is normally set by the `FastAPI` app during the call to `app.include_router(some_router)`.
+
+ But as we are never calling `app.include_router(some_router)`, we need to set the `default_response_class` during creation of the `APIRouter`.
+
+```Python hl_lines="3 24"
+{!./src/openapi_callbacks/tutorial001.py!}
+```
+
+### Create the callback *path operation*
+
+To create the callback *path operation* use the same `APIRouter` you created above.
+
+It should look just like a normal FastAPI *path operation*:
+
+* It should probably have a declaration of the body it should receive, e.g. `body: InvoiceEvent`.
+* And it could also have a declaration of the response it should return, e.g. `response_model=InvoiceEventReceived`.
+
+```Python hl_lines="15 16 17 20 21 27 28 29 30 31 32"
+{!./src/openapi_callbacks/tutorial001.py!}
+```
+
+There are 2 main differences from a normal *path operation*:
+
+* It doesn't need to have any actual code, because your app will never call this code. It's only used to document the *external API*. So, the function could just have `pass`.
+* The *path* can contain an <a href="https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.2.md#key-expression" target="_blank">OpenAPI 3 expression</a> (see more below) where it can use variables with parameters and parts of the original request sent to *your API*.
+
+### The callback path expression
+
+The callback *path* can have an <a href="https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.2.md#key-expression" target="_blank">OpenAPI 3 expression</a> that can contain parts of the original request sent to *your API*.
+
+In this case, it's the `str`:
+
+```Python
+"{$callback_url}/invoices/{$request.body.id}"
+```
+
+So, if your API user (the external developer) sends a request to *your API* to:
+
+```
+https://yourapi.com/invoices/?callback_url=https://www.external.org/events
+```
+
+with a JSON body of:
+
+```JSON
+{
+ "id": "2expen51ve",
+ "customer": "Mr. Richie Rich",
+ "total": "9999"
+}
+```
+
+Then *your API* will process the invoice, and at some point later, send a callback request to the `callback_url` (the *external API*):
+
+```
+https://www.external.org/events/invoices/2expen51ve
+```
+
+with a JSON body containing something like:
+
+```JSON
+{
+ "description": "Payment celebration",
+ "paid": true
+}
+```
+
+and it would expect a response from that *external API* with a JSON body like:
+
+```JSON
+{
+ "ok": true
+}
+```
+
+!!! tip
+ Notice how the callback URL used contains the URL received as a query parameter in `callback_url` (`https://www.external.org/events`) and also the invoice `id` from inside of the JSON body (`2expen51ve`).
+
+### Add the callback router
+
+At this point you have the *callback path operation(s)* needed (the one(s) that the *external developer* should implement in the *external API*) in the callback router you created above.
+
+Now use the parameter `callbacks` in *your API's path operation decorator* to pass the attribute `.routes` (that's actually just a `list` of routes/*path operations*) from that callback router:
+
+```Python hl_lines="35"
+{!./src/openapi_callbacks/tutorial001.py!}
+```
+
+!!! tip
+ Notice that you are not passing the router itself (`invoices_callback_router`) to `callback=`, but the attribute `.routes`, as in `invoices_callback_router.routes`.
+
+### Check the docs
+
+Now you can start your app with Uvicorn and go to <a href="http://127.0.0.1:8000/docs" target="_blank">http://127.0.0.1:8000/docs</a>.
+
+You will see your docs including a "Callback" section for your *path operation* that shows how the *external API* should look like:
+
+<img src="/img/tutorial/openapi-callbacks/image01.png">
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[routing.APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
include_in_schema=include_in_schema,
response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def put(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[routing.APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
include_in_schema=include_in_schema,
response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def post(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[routing.APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
include_in_schema=include_in_schema,
response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def delete(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[routing.APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
include_in_schema=include_in_schema,
response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def options(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[routing.APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
include_in_schema=include_in_schema,
response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def head(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[routing.APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
include_in_schema=include_in_schema,
response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def patch(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[routing.APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
include_in_schema=include_in_schema,
response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def trace(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[routing.APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
include_in_schema=include_in_schema,
response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
)
if request_body_oai:
operation["requestBody"] = request_body_oai
+ if route.callbacks:
+ callbacks = {}
+ for callback in route.callbacks:
+ cb_path, cb_security_schemes, cb_definitions, = get_openapi_path(
+ route=callback, model_name_map=model_name_map
+ )
+ callbacks[callback.name] = {callback.path: cb_path}
+ operation["callbacks"] = callbacks
if route.responses:
for (additional_status_code, response) in route.responses.items():
assert isinstance(
include_in_schema: bool = True,
response_class: Optional[Type[Response]] = None,
dependency_overrides_provider: Any = None,
+ callbacks: Optional[List["APIRoute"]] = None,
) -> None:
self.path = path
self.endpoint = endpoint
)
self.body_field = get_body_field(dependant=self.dependant, name=self.unique_id)
self.dependency_overrides_provider = dependency_overrides_provider
+ self.callbacks = callbacks
self.app = request_response(self.get_route_handler())
def get_route_handler(self) -> Callable:
default: ASGIApp = None,
dependency_overrides_provider: Any = None,
route_class: Type[APIRoute] = APIRoute,
+ default_response_class: Type[Response] = None,
) -> None:
super().__init__(
routes=routes, redirect_slashes=redirect_slashes, default=default
)
self.dependency_overrides_provider = dependency_overrides_provider
self.route_class = route_class
+ self.default_response_class = default_response_class
def add_api_route(
self,
response_class: Type[Response] = None,
name: str = None,
route_class_override: Optional[Type[APIRoute]] = None,
+ callbacks: List[APIRoute] = None,
) -> None:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
response_model_exclude_unset or response_model_skip_defaults
),
include_in_schema=include_in_schema,
- response_class=response_class,
+ response_class=response_class or self.default_response_class,
name=name,
dependency_overrides_provider=self.dependency_overrides_provider,
+ callbacks=callbacks,
)
self.routes.append(route)
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
response_model_exclude_unset or response_model_skip_defaults
),
include_in_schema=include_in_schema,
- response_class=response_class,
+ response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
return func
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
response_model_exclude_unset or response_model_skip_defaults
),
include_in_schema=include_in_schema,
- response_class=response_class,
+ response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def put(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
response_model_exclude_unset or response_model_skip_defaults
),
include_in_schema=include_in_schema,
- response_class=response_class,
+ response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def post(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
response_model_exclude_unset or response_model_skip_defaults
),
include_in_schema=include_in_schema,
- response_class=response_class,
+ response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def delete(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
response_model_exclude_unset or response_model_skip_defaults
),
include_in_schema=include_in_schema,
- response_class=response_class,
+ response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def options(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
response_model_exclude_unset or response_model_skip_defaults
),
include_in_schema=include_in_schema,
- response_class=response_class,
+ response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def head(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
response_model_exclude_unset or response_model_skip_defaults
),
include_in_schema=include_in_schema,
- response_class=response_class,
+ response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def patch(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
response_model_exclude_unset or response_model_skip_defaults
),
include_in_schema=include_in_schema,
- response_class=response_class,
+ response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def trace(
include_in_schema: bool = True,
response_class: Type[Response] = None,
name: str = None,
+ callbacks: List[APIRoute] = None,
) -> Callable:
if response_model_skip_defaults is not None:
warning_response_model_skip_defaults_deprecated() # pragma: nocover
response_model_exclude_unset or response_model_skip_defaults
),
include_in_schema=include_in_schema,
- response_class=response_class,
+ response_class=response_class or self.default_response_class,
name=name,
+ callbacks=callbacks,
)
def get_flat_models_from_routes(routes: Sequence[BaseRoute]) -> Set[Type[BaseModel]]:
body_fields_from_routes: List[ModelField] = []
responses_from_routes: List[ModelField] = []
+ callback_flat_models: Set[Type[BaseModel]] = set()
for route in routes:
if getattr(route, "include_in_schema", None) and isinstance(
route, routing.APIRoute
responses_from_routes.append(route.response_field)
if route.response_fields:
responses_from_routes.extend(route.response_fields.values())
- flat_models = get_flat_models_from_fields(
+ if route.callbacks:
+ callback_flat_models |= get_flat_models_from_routes(route.callbacks)
+ flat_models = callback_flat_models | get_flat_models_from_fields(
body_fields_from_routes + responses_from_routes, known_models=set()
)
return flat_models
def generate_operation_id_for_path(*, name: str, path: str, method: str) -> str:
operation_id = name + path
- operation_id = operation_id.replace("{", "_").replace("}", "_").replace("/", "_")
+ operation_id = re.sub("[^0-9a-zA-Z_]", "_", operation_id)
operation_id = operation_id + "_" + method.lower()
return operation_id
- Testing Dependencies with Overrides: 'tutorial/testing-dependencies.md'
- Debugging: 'tutorial/debugging.md'
- Extending OpenAPI: 'tutorial/extending-openapi.md'
+ - OpenAPI Callbacks: 'tutorial/openapi-callbacks.md'
- Concurrency and async / await: 'async.md'
- Deployment: 'deployment.md'
- Project Generation - Template: 'project-generation.md'
},
},
"summary": "Get Path Param Required Id",
- "operationId": "get_path_param_required_id_path_param-required__item_id__get",
+ "operationId": "get_path_param_required_id_path_param_required__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Min Length",
- "operationId": "get_path_param_min_length_path_param-minlength__item_id__get",
+ "operationId": "get_path_param_min_length_path_param_minlength__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Max Length",
- "operationId": "get_path_param_max_length_path_param-maxlength__item_id__get",
+ "operationId": "get_path_param_max_length_path_param_maxlength__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Min Max Length",
- "operationId": "get_path_param_min_max_length_path_param-min_maxlength__item_id__get",
+ "operationId": "get_path_param_min_max_length_path_param_min_maxlength__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Gt",
- "operationId": "get_path_param_gt_path_param-gt__item_id__get",
+ "operationId": "get_path_param_gt_path_param_gt__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Gt0",
- "operationId": "get_path_param_gt0_path_param-gt0__item_id__get",
+ "operationId": "get_path_param_gt0_path_param_gt0__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Ge",
- "operationId": "get_path_param_ge_path_param-ge__item_id__get",
+ "operationId": "get_path_param_ge_path_param_ge__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Lt",
- "operationId": "get_path_param_lt_path_param-lt__item_id__get",
+ "operationId": "get_path_param_lt_path_param_lt__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Lt0",
- "operationId": "get_path_param_lt0_path_param-lt0__item_id__get",
+ "operationId": "get_path_param_lt0_path_param_lt0__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Le",
- "operationId": "get_path_param_le_path_param-le__item_id__get",
+ "operationId": "get_path_param_le_path_param_le__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Lt Gt",
- "operationId": "get_path_param_lt_gt_path_param-lt-gt__item_id__get",
+ "operationId": "get_path_param_lt_gt_path_param_lt_gt__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Le Ge",
- "operationId": "get_path_param_le_ge_path_param-le-ge__item_id__get",
+ "operationId": "get_path_param_le_ge_path_param_le_ge__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Lt Int",
- "operationId": "get_path_param_lt_int_path_param-lt-int__item_id__get",
+ "operationId": "get_path_param_lt_int_path_param_lt_int__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Gt Int",
- "operationId": "get_path_param_gt_int_path_param-gt-int__item_id__get",
+ "operationId": "get_path_param_gt_int_path_param_gt_int__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Le Int",
- "operationId": "get_path_param_le_int_path_param-le-int__item_id__get",
+ "operationId": "get_path_param_le_int_path_param_le_int__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Ge Int",
- "operationId": "get_path_param_ge_int_path_param-ge-int__item_id__get",
+ "operationId": "get_path_param_ge_int_path_param_ge_int__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Lt Gt Int",
- "operationId": "get_path_param_lt_gt_int_path_param-lt-gt-int__item_id__get",
+ "operationId": "get_path_param_lt_gt_int_path_param_lt_gt_int__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Path Param Le Ge Int",
- "operationId": "get_path_param_le_ge_int_path_param-le-ge-int__item_id__get",
+ "operationId": "get_path_param_le_ge_int_path_param_le_ge_int__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Query Param Required",
- "operationId": "get_query_param_required_query_param-required_get",
+ "operationId": "get_query_param_required_query_param_required_get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Query Param Required Type",
- "operationId": "get_query_param_required_type_query_param-required_int_get",
+ "operationId": "get_query_param_required_type_query_param_required_int_get",
"parameters": [
{
"required": True,
},
},
"summary": "Get Not Decorated",
- "operationId": "get_not_decorated_items-not-decorated__item_id__get",
+ "operationId": "get_not_decorated_items_not_decorated__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Create Item",
- "operationId": "create_item_starlette-items__item_id__get",
+ "operationId": "create_item_starlette_items__item_id__get",
"parameters": [
{
"required": True,
},
},
"summary": "Create Index Weights",
- "operationId": "create_index_weights_index-weights__post",
+ "operationId": "create_index_weights_index_weights__post",
"requestBody": {
"content": {
"application/json": {
"content": {
"application/json": {
"schema": {
- "title": "Response Read Keyword Weights Keyword-Weights Get",
+ "title": "Response Read Keyword Weights Keyword Weights Get",
"type": "object",
"additionalProperties": {"type": "number"},
}
}
},
"summary": "Read Keyword Weights",
- "operationId": "read_keyword_weights_keyword-weights__get",
+ "operationId": "read_keyword_weights_keyword_weights__get",
}
}
},
},
},
"summary": "Read Item Header",
- "operationId": "read_item_header_items-header__item_id__get",
+ "operationId": "read_item_header_items_header__item_id__get",
"parameters": [
{
"required": True,
--- /dev/null
+from starlette.testclient import TestClient
+
+from openapi_callbacks.tutorial001 import app, invoice_notification
+
+client = TestClient(app)
+
+openapi_schema = {
+ "openapi": "3.0.2",
+ "info": {"title": "Fast API", "version": "0.1.0"},
+ "paths": {
+ "/invoices/": {
+ "post": {
+ "summary": "Create Invoice",
+ "description": 'Create an invoice.\n\nThis will (let\'s imagine) let the API user (some external developer) create an\ninvoice.\n\nAnd this path operation will:\n\n* Send the invoice to the client.\n* Collect the money from the client.\n* Send a notification back to the API user (the external developer), as a callback.\n * At this point is that the API will somehow send a POST request to the\n external API with the notification of the invoice event\n (e.g. "payment successful").',
+ "operationId": "create_invoice_invoices__post",
+ "parameters": [
+ {
+ "required": False,
+ "schema": {
+ "title": "Callback Url",
+ "maxLength": 2083,
+ "minLength": 1,
+ "type": "string",
+ "format": "uri",
+ },
+ "name": "callback_url",
+ "in": "query",
+ }
+ ],
+ "requestBody": {
+ "content": {
+ "application/json": {
+ "schema": {"$ref": "#/components/schemas/Invoice"}
+ }
+ },
+ "required": True,
+ },
+ "responses": {
+ "200": {
+ "description": "Successful Response",
+ "content": {"application/json": {"schema": {}}},
+ },
+ "422": {
+ "description": "Validation Error",
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError"
+ }
+ }
+ },
+ },
+ },
+ "callbacks": {
+ "invoice_notification": {
+ "{$callback_url}/invoices/{$request.body.id}": {
+ "post": {
+ "summary": "Invoice Notification",
+ "operationId": "invoice_notification__callback_url__invoices___request_body_id__post",
+ "requestBody": {
+ "required": True,
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/InvoiceEvent"
+ }
+ }
+ },
+ },
+ "responses": {
+ "200": {
+ "description": "Successful Response",
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/InvoiceEventReceived"
+ }
+ }
+ },
+ },
+ "422": {
+ "description": "Validation Error",
+ "content": {
+ "application/json": {
+ "schema": {
+ "$ref": "#/components/schemas/HTTPValidationError"
+ }
+ }
+ },
+ },
+ },
+ }
+ }
+ }
+ },
+ }
+ }
+ },
+ "components": {
+ "schemas": {
+ "HTTPValidationError": {
+ "title": "HTTPValidationError",
+ "type": "object",
+ "properties": {
+ "detail": {
+ "title": "Detail",
+ "type": "array",
+ "items": {"$ref": "#/components/schemas/ValidationError"},
+ }
+ },
+ },
+ "Invoice": {
+ "title": "Invoice",
+ "required": ["id", "customer", "total"],
+ "type": "object",
+ "properties": {
+ "id": {"title": "Id", "type": "string"},
+ "title": {"title": "Title", "type": "string"},
+ "customer": {"title": "Customer", "type": "string"},
+ "total": {"title": "Total", "type": "number"},
+ },
+ },
+ "InvoiceEvent": {
+ "title": "InvoiceEvent",
+ "required": ["description", "paid"],
+ "type": "object",
+ "properties": {
+ "description": {"title": "Description", "type": "string"},
+ "paid": {"title": "Paid", "type": "boolean"},
+ },
+ },
+ "InvoiceEventReceived": {
+ "title": "InvoiceEventReceived",
+ "required": ["ok"],
+ "type": "object",
+ "properties": {"ok": {"title": "Ok", "type": "boolean"}},
+ },
+ "ValidationError": {
+ "title": "ValidationError",
+ "required": ["loc", "msg", "type"],
+ "type": "object",
+ "properties": {
+ "loc": {
+ "title": "Location",
+ "type": "array",
+ "items": {"type": "string"},
+ },
+ "msg": {"title": "Message", "type": "string"},
+ "type": {"title": "Error Type", "type": "string"},
+ },
+ },
+ }
+ },
+}
+
+
+def test_openapi():
+ with client:
+ response = client.get("/openapi.json")
+
+ assert response.json() == openapi_schema
+
+
+def test_get():
+ response = client.post(
+ "/invoices/", json={"id": "fooinvoice", "customer": "John", "total": 5.3}
+ )
+ assert response.status_code == 200
+ assert response.json() == {"msg": "Invoice received"}
+
+
+def test_dummy_callback():
+ # Just for coverage
+ invoice_notification({})
projects / thirdparty / fastapi / fastapi.git / commitdiff
