Web Server¶

In Superdesk v3.0+ we use an abstraction layer when it comes to the web server functionality. This means that we don’t directly interact with the Flask library, but Superdesk specific classes and functionality instead.

For example:

from superdesk.core.module import Module
from superdesk.core.web import (
    endpoint,
    Request,
    Response,
)

@endpoint("hello/world", methods=["GET"])
async def hello_world(request: Request) -> Response:
    return Response(
        body="hello, world!",
        status_code=200,
        headers=()
    )

module = Module(
    name="tests.http",
    endpoints=[hello_world]
)

Declaring Endpoints¶

There are three ways to declare a function as an HTTP endpoint.

Using the endpoint decorator:

from superdesk.core.web import (
    Request,
    Response,
    endpoint,
)

@endpoint("hello/world", methods=["GET"])
async def hello_world(request: Request) -> Response:
    ...

Using the Endpoint class:

from superdesk.core.web import (
    Request,
    Response,
    Endpoint,
)

async def hello_world(request: Request) -> Response:
    ...

endpoint = Endpoint(
    url="hello/world",
    methods=["GET"],
    func=hello_world
)

Using the EndpointGroup class group:

from superdesk.core.web import (
    Request,
    Response,
    EndpointGroup,
)

group = EndpointGroup(
    name="tests",
    import_name=__name__,
    url_prefix="hello"
)

@group.endpoint(url="world", methods=["GET"])
async def hello_world(request: Request) -> Response:
    ...

Registering Endpoints:¶

There are two ways to register an endpoint with the system.

Manually using the WSGIApp.register_endpoint function:

from superdesk.core.app import SuperdeskAsyncApp

def init(app: SuperdeskAsyncApp):
    app.wsgi.register_endpoint(hello_world)

Automatically with the endpoints module config:

from superdesk.core.module import Module

module = Module(
    name="my.module",
    endpoints=[hello_world, group]
)

Resource REST Endpoints¶

REST endpoints can be enabled for a resource by defining the rest_endpoints attribute on the ResourceConfig. See RestEndpointConfig for config options.

For example:

from superdesk.core.module import Module
from superdesk.core.resources import (
    ResourceConfig,
    ResourceModel,
    RestEndpointConfig,
)

# Define your resource model and config
class User(ResourceModel):
    first_name: str
    last_name: str

# Configure the resource
user_resource_config = ResourceConfig(
    name="users",
    data_class=User,

    # Including the `rest_endpoints` config
    rest_endpoints=RestEndpointConfig(),
)

module = Module(
    name="tests.users",
    resources=[user_resource_config],
)

Resource REST Endpoint with Parent¶

REST endpoints can also include a parent/child relationship with the resource. This is achieved using the RestParentLink attribute on the RestEndpointConfig.

Example config:

from typing import Annotated
from superdesk.core.module import Module
from superdesk.core.resources import (
    ResourceConfig,
    ResourceModel,
    RestEndpointConfig,
    RestParentLink,
)
from superdesk.core.resources.validators import (
    validate_data_relation_async,
)

# 1. Define parent resource and config
class Company(ResourceModel):
    name: str

company_resource_config = ResourceConfig(
    name="companies",
    data_class=Company,
    rest_endpoints=RestEndpointConfig()
)

# 2. Define child resource and config
class User(ResourceModel):
    first_name: str
    last_name: str

    # 2a. Include a field that references the parent
    company: Annotated[
        str,
        validate_data_relation_async(
            company_resource_config.name,
        ),
    ]

user_resource_config = ResourceConfig(
    name="users",
    data_class=User,
    rest_endpoints=RestEndpointConfig(

        # 2b. Include a link to Company as a parent resource
        parent_links=[
            RestParentLink(
                resource_name=company_resource_config.name,
                model_id_field="company",
            ),
        ],
    ),
)

# 3. Register the resources with a module
module = Module(
    name="tests.users",
    resources=[
        company_resource_config,
        user_resource_config,
    ],
)

The above example exposes the following URLs:

/api/companies
/api/companies/<item_id>
/api/companies/<company>/users
/api/companies/<company>/users/<item_id>

As you can see the users endpoints are prefixed with /api/company/<company>/.

This provides the following functionality:

Validation that a Company must exist for the user
Populates the company field of a User with the ID from the URL
When searching for users, will only provide users for the specific company provided in the URL of the request

For example:

async def test_users():
    # Create the parent Company
    response = await client.post(
        "/api/company",
        json={"name": "Sourcefabric"}
    )

    # Retrieve the Company ID from the response
    company_id = (await response.get_json())[0]

    # Attemps to create a user with non-existing company
    # responds with a 404 - NotFound error
    response = await client.post(
        f"/api/company/blah_blah/users",
        json={"first_name": "Monkey", "last_name": "Mania"}
    )
    assert response.status_code == 404

    # Create the new User
    # Notice the ``company_id`` is used in the URL
    response = await client.post(
        f"/api/company/{company_id}/users",
        json={"first_name": "Monkey", "last_name": "Mania"}
    )
    user_id = (await response.get_json())[0]

    # Retrieve the new user
    response = await client.get(
        f"/api/company/{company_id}/users/{user_id}"
    )
    user_dict = await response.get_json()
    assert user_dict["company"] == company_id

    # Retrieve all company users
    response = await client.get(
        f"/api/company/{company_id}/users"
    )
    users_dict = (await response.get_json())["_items"]
    assert len(users_dict) == 1
    assert users_dict[0]["_id"] == user_id

Validation¶

Request route arguments and URL params can be validated against Pydantic models. All you need to do is to define the model for each argument type, and the system will validate them when processing the request. If the request does not pass validation, a Pydantic ValidationError will be raised.

For example:

from typing import Optional
from enum import Enum
from pydantic import BaseModel

class UserActions(str, Enum):
    activate = "activate"
    disable = "disable"

class RouteArguments(BaseModel):
    user_id: str
    action: UserActions

class URLParams(BaseModel):
    verbose: bool = False

@endpoint(
    "users_async/<string:user_id>/action/<string:action>",
    methods=["GET"]
)
async def hello_world(
    args: RouteArguments,
    params: URLParams,
    request: Request,
) -> Response:
    # If the request reaches this line,
    # we have valid arguments & params
    user = get_user(args.user_id)
    if args.action == UserActions.activate:
        pass
    elif args.action == UserActions.disable:
        pass
    else:
        # This line should never be reached,
        # as validation would have caught this already
        assert False, "unreachable"

    return Response(
        "hello, world!",
        200,
        ()
    )

def init(app: SuperdeskAsyncApp) -> None:
    app.wsgi.register_endpoint(hello_world)

API References¶

class Response(body: Any, status_code: int = 200, headers: Sequence = ())[source]¶

Dataclass for endpoints to return response from a request

body: Any¶: The body of the response (Flask will determine data type for us)

status_code: int = 200¶: HTTP Status Code of the response

headers: Sequence = ()¶: Any additional headers to be added

EndpointFunction[source]¶: alias of Callable[[], Awaitable[Response]] | Callable[[Request], Awaitable[Response]] | Callable[[PydanticModelType, PydanticModelType, Request], Awaitable[Response]] | Callable[[None, PydanticModelType, Request], Awaitable[Response]] | Callable[[PydanticModelType, None, Request], Awaitable[Response]] | Callable[[None, None, Request], Awaitable[Response]] | Callable[[], Response] | Callable[[Request], Response] | Callable[[PydanticModelType, PydanticModelType, Request], Response] | Callable[[None, PydanticModelType, Request], Response] | Callable[[PydanticModelType, None, Request], Response] | Callable[[None, None, Request], Response]

class Endpoint(url: str, func: Callable[[], Awaitable[Response]] | Callable[[Request], Awaitable[Response]] | Callable[[PydanticModelType, PydanticModelType, Request], Awaitable[Response]] | Callable[[None, PydanticModelType, Request], Awaitable[Response]] | Callable[[PydanticModelType, None, Request], Awaitable[Response]] | Callable[[None, None, Request], Awaitable[Response]] | Callable[[], Response] | Callable[[Request], Response] | Callable[[PydanticModelType, PydanticModelType, Request], Response] | Callable[[None, PydanticModelType, Request], Response] | Callable[[PydanticModelType, None, Request], Response] | Callable[[None, None, Request], Response], methods: list[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']] | None = None, name: str | None = None, title: str | None = None, auth: Literal[False] | list[Callable[[Request], Awaitable[Any | None]]] | dict[str, Callable[[Request], Awaitable[Any | None]]] | None = None, parent: EndpointGroup | None = None, cors: bool | None = None, tags: list[str] | None = None, summary: str | None = None, description: str | None = None, responses: dict | None = None)[source]¶

Base class used for registering and processing endpoints

url: str¶: URL for the endpoint If the URL starts with “/”, it will be used as-is. Otherwise, the API prefix will be prepended to the URL. For example: - “items” -> “{api_prefix}{api_version}/items” - “/custom/path” -> “/custom/path”

func: Callable[[], Awaitable[Response]] | Callable[[Request], Awaitable[Response]] | Callable[[PydanticModelType, PydanticModelType, Request], Awaitable[Response]] | Callable[[None, PydanticModelType, Request], Awaitable[Response]] | Callable[[PydanticModelType, None, Request], Awaitable[Response]] | Callable[[None, None, Request], Awaitable[Response]] | Callable[[], Response] | Callable[[Request], Response] | Callable[[PydanticModelType, PydanticModelType, Request], Response] | Callable[[None, PydanticModelType, Request], Response] | Callable[[PydanticModelType, None, Request], Response] | Callable[[None, None, Request], Response]¶: The callback function used to process the request

methods: list[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']]¶: HTTP Methods allowed for this endpoint

name: str¶: Name of the endpoint (must be unique)

title: str¶: Title of the endpoint (used for HATEOAS links)

auth: Literal[False] | list[Callable[[Request], Awaitable[Any | None]]] | dict[str, Callable[[Request], Awaitable[Any | None]]] | None¶

parent: EndpointGroup | None¶

cors: bool | None¶: If True, CORS will be enabled for this endpoint (default is in ASYNC_ENABLE_CORS config setting).

tags: list[str]¶: Optional tags (for use with OpenAPI documentation)

summary: str | None¶: Optional summary (for use with OpenAPI documentation)

description: str | None¶: Optional description (for use with OpenAPI documentation)

responses: dict[str, dict] | None¶: Optional responses (for use with OpenAPI documentation)

get_auth_rules() → Literal[False] | list[Callable[[Request], Awaitable[Any | None]]] | dict[str, Callable[[Request], Awaitable[Any | None]]] | None¶

property add_cors_headers: bool¶

class Request(*args, **kwargs)[source]¶

Protocol to define common request functionality

This is implemented in SuperdeskEve app using Flask to provide the required functionality.

endpoint: Endpoint¶: The current Endpoint being processed

storage: RequestStorage¶

user: Any | None¶

property method: Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']¶: Returns the current HTTP method for the request

property url: str¶: Returns the URL of the current request

property path: str¶: Returns the URL of the current request

get_header(key: str) → str | None¶: Get an HTTP header from the current request

async get_json(force: bool = False) → Any | None¶: Get the body of the current request in JSON format

async get_form() → Mapping¶: Get the body of the current request in form format

async get_data() → bytes | str¶: Get the body of the current request in raw bytes format

async abort(code: int, *args: Any, **kwargs: Any) → NoReturn¶

get_view_args(key: str) → str | None¶

get_url_arg(key: str) → str | None¶

redirect(location: str, code: int = 302) → Any¶

is_json_request() → bool¶

class EndpointGroup(url: str, func: Callable[[], Awaitable[Response]] | Callable[[Request], Awaitable[Response]] | Callable[[PydanticModelType, PydanticModelType, Request], Awaitable[Response]] | Callable[[None, PydanticModelType, Request], Awaitable[Response]] | Callable[[PydanticModelType, None, Request], Awaitable[Response]] | Callable[[None, None, Request], Awaitable[Response]] | Callable[[], Response] | Callable[[Request], Response] | Callable[[PydanticModelType, PydanticModelType, Request], Response] | Callable[[None, PydanticModelType, Request], Response] | Callable[[PydanticModelType, None, Request], Response] | Callable[[None, None, Request], Response], methods: list[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']] | None = None, name: str | None = None, title: str | None = None, auth: Literal[False] | list[Callable[[Request], Awaitable[Any | None]]] | dict[str, Callable[[Request], Awaitable[Any | None]]] | None = None, parent: EndpointGroup | None = None, cors: bool | None = None, tags: list[str] | None = None, summary: str | None = None, description: str | None = None, responses: dict | None = None)[source]¶

Base class used for registering a group of endpoints

import_name: str¶: The import name of the module where this object is defined. Usually __name__ should be used.

url_prefix: str | None¶: Optional url prefix to be added to all routes of this group

endpoints: list[Endpoint]¶: List of endpoints registered with this group

endpoint(url: str, name: str | None = None, title: str | None = None, methods: list[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']] | None = None, auth: Literal[False] | list[Callable[[Request], Awaitable[Any | None]]] | dict[str, Callable[[Request], Awaitable[Any | None]]] | None = None)¶

class RestResponseMeta[source]¶

Dictionary to hold the response metadata for a REST request

page: int¶: Current page requested

max_results: int¶: Maximum results requested

total: int¶: Total number of documents found

class RestGetResponse[source]¶: Dictionary to hold the response for a REST request

Decorator function to convert a pure function to an Endpoint instance which is later used to register with a Module or the app.

Parameters:

url – The URL of the endpoint. If it starts with “/”, it will be used exactly as provided. Otherwise, the API prefix and version will be automatically prepended. For example: - “items” becomes “{api_prefix}{api_version}/items” - “/custom/path” stays as “/custom/path”
name – The optional name of the endpoint
title – The optional title of the endpoint (used for HATEOAS links)
methods – The optional list of HTTP methods allowed
auth – The auth configuration for this endpoint
cors – If True, CORS will be enabled for this endpoint. If None, the value will be taken from the ASYNC_ENABLE_CORS config setting.

class RestEndpoints(url: str, name: str, import_name: str | None = None, resource_methods: list[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']] | None = None, item_methods: list[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']] | None = None, id_param_type: str | None = None, auth: Literal[False] | list[Callable[[Request], Awaitable[Any | None]]] | dict[str, Callable[[Request], Awaitable[Any | None]]] | None = None)[source]¶

Custom EndpointGroup for REST resources

auth: Literal[False] | list[Callable[[Request], Awaitable[Any | None]]] | dict[str, Callable[[Request], Awaitable[Any | None]]] | None = None¶

url: str¶: The URL for this resource

resource_methods: list[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']]¶: The list of HTTP methods for the resource endpoints

item_methods: list[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']]¶: The list of HTTP methods for the resource item endpoints

id_param_type: str¶: Optionally set the route param type for the ID, defaults to string

add_endpoints()¶

get_resource_url() → str¶: Returns the URL for this resource, for registering with WSGI

get_item_url(arg_name: str = 'item_id') → str¶

Returns the URL for an item of this resource, for registering with WSGI

Parameters:: arg_name – The name of the URL argument to use for the resource item URL
Returns:: The URL for an item of this resource

gen_url_for_item(request: Request, item_id: str | ObjectId, force_append_id: bool = False) → str¶

Get the URL of an item of this resource, for HATEOAS self link

Parameters:

request – The request instance currently being processed
item_id – The ID of the item being returned
force_append_id – If True, appends the item_id to the path if doesn’t end with it

Returns:

The URL of an item of this resource

async get_item(args: ItemRequestViewArgs, params: Any, request: Request) → Response¶: Processes a get single item request

async create_item(request: Request) → Response¶: Processes a create item request

async update_item(args: ItemRequestViewArgs, params: None, request: Request) → Response¶: Processes an update item request

async delete_item(args: ItemRequestViewArgs, params: None, request: Request) → Response¶: Processes a delete item request

async search_items(args: None, params: SearchRequest, request: Request) → Response¶: Processes a search request

class RestEndpointConfig(resource_methods: list[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']] | None = None, item_methods: list[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']] | None = None, endpoints_class: type['ResourceRestEndpoints'] | None = None, id_param_type: str | None = None, url: str | None = None, parent_links: list[superdesk.core.resources.resource_rest_endpoints.RestParentLink] | None = None, auth: Literal[False] | list[Callable[[superdesk.core.types.web.Request], Awaitable[Any | None]]] | dict[str, Callable[[superdesk.core.types.web.Request], Awaitable[Any | None]]] | NoneType = None, enable_cors: bool | None = None, populate_item_hateoas: bool | None = None, exclude_fields_in_response: list[str] | dict[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS'], list[str]] | None = None, additional_lookup: superdesk.core.resources.resource_rest_endpoints.AdditionalLookupConfig | None = None)[source]¶

resource_methods: list[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']] | None = None¶: Optional list of resource level methods, defaults to [“GET”, “POST”]

item_methods: list[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']] | None = None¶: Optional list of item level methods, defaults to [“GET”, “PATCH”, “DELETE”]

endpoints_class: type[ResourceRestEndpoints] | None = None¶: Optional EndpointGroup, will default to ResourceRestEndpoints

id_param_type: str | None = None¶: Optionally set a custom URL ID param syntax for item routes

url: str | None = None¶: Optionally set a custom URL for routes, defaults to resource_name

parent_links: list[RestParentLink] | None = None¶: Optionally assign parent resource(s) for this resource (parent/child relationship) This will prepend this resources URL with the URL of the parent resource item

auth: Literal[False] | list[Callable[[Request], Awaitable[Any | None]]] | dict[str, Callable[[Request], Awaitable[Any | None]]] | None = None¶

enable_cors: bool | None = None¶

populate_item_hateoas: bool | None = None¶

exclude_fields_in_response: list[str] | dict[Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS'], list[str]] | None = None¶: Exclude these fields from ALL responses (uses MongoDB or Elasticsearch to provide projection) Raises an exception if client specifically requests any of these fields to be included

additional_lookup: AdditionalLookupConfig | None = None¶

class RestParentLink(resource_name: str, model_id_field: str | None = None, url_arg_name: str | None = None, parent_id_field: str = '_id')[source]¶

resource_name: str¶: Name of the resource this parent link belongs to

model_id_field: str | None = None¶: Field used to store the resource ID in the child resource, defaults to resource_name

url_arg_name: str | None = None¶: Name of the URL argument in the route, defaults to model_id_field

parent_id_field: str = '_id'¶: ID Field of the parent used when searching for parent item resource, defaults to model_id_field

get_model_id_field() → str¶: Get the ID Field for the local model used to store the reference to the parent model

get_url_arg_name() → str¶: Get the name of hte URL argument used in the route

class ResourceRestEndpoints(resource_config: ResourceConfig, endpoint_config: RestEndpointConfig)[source]¶

Custom EndpointGroup for REST resources

resource_config: ResourceConfig¶: The config for the resource to use

endpoint_config: RestEndpointConfig¶: REST endpoint config

add_endpoints()¶

get_resource_url() → str¶

Returns the URL for this resource, for registering with WSGI

If the resource has parent_links configured, these will be used to construct the URL with the parent resources URL and item ID

get_item_url(arg_name: str = 'item_id') → str¶

Returns the URL for an item of this resource, for registering with WSGI

Parameters:: arg_name – The name of the URL argument to use for the resource item URL
Returns:: The URL for an item of this resource

async get_parent_items(request: Request) → dict[str, dict]¶

Returns a dictionary of resource name to item for configured parent links

Returns:: A dictionary, with the key being the resource name and value being the parent item
Raises:: SuperdeskApiError.badRequestError – If a parent item is not found

construct_parent_item_lookup(request: Request) → dict¶

Prefills a MongoDB query with the parent attributes from the request

This is used to filter items of this resource to make sure they belong to all parent item(s).

Parameters:: request – The request object currently being processed
Returns:: A MongoDB query

property service¶

get_exclude_fields_projection(method: Literal['GET', 'POST', 'PATCH', 'PUT', 'DELETE', 'HEAD', 'OPTIONS']) → dict[str, Literal[False]] | None¶

async get_item(args: ItemRequestViewArgs, params: ItemRequestUrlArgs, request: Request) → Response¶: Processes a get single item request

async get_item_from_additional_lookup(args: dict, params: ItemRequestUrlArgs, request: Request) → Response¶: Processes a get single item request, using additional lookup config to get the item

async create_item(request: Request) → Response¶: Processes a create item request

async update_item(args: ItemRequestViewArgs, params: None, request: Request) → Response¶: Processes an update item request

async delete_item(args: ItemRequestViewArgs, params: None, request: Request) → Response¶: Processes a delete item request

update_where_filter(params: SearchRequest, where: dict)¶

async search_items(args: None, params: SearchRequest, request: Request) → Response¶: Processes a search request

async resource_options_endpoint() → Response¶

async item_options_endpoint() → Response¶

get_resource_cors_headers()¶

get_item_cors_headers()¶

async on_fetched_item(request: Request, doc: dict) → None¶

async on_fetched(request: Request, doc: RestGetResponse) → None¶

class WSGIApp(*args, **kwargs)[source]¶

Protocol for defining functionality from a WSGI application (such as Eve/Flask)

A class instance that adheres to this protocol is passed into the SuperdeskAsyncApp constructor. This way the SuperdeskAsyncApp does not need to know the underlying WSGI application, just that it provides certain functionality.

config: dict[str, Any]¶: Config for the application

client_config: dict[str, Any]¶: Config for the front-end application

testing: bool = False¶

media: Any¶: Interface to upload/download/query media

mail: Any¶

data: Any¶

storage: Any¶

auth: Any¶

subjects: Any¶

notification_client: NotificationClientProtocol¶

locators: Any¶

celery: Any¶

redis: Any¶

jinja_loader: Any¶

jinja_env: Any¶

extensions: dict[str, Any]¶

register_endpoint(endpoint: Endpoint | EndpointGroup)¶

register_resource(name: str, settings: dict[str, Any])¶

upload_url(media_id: str) → str¶

download_url(media_id: str) → str¶

app_context()¶

get_current_user_dict() → dict[str, Any] | None¶

response_class(*args, **kwargs) → Any¶

validator(*args, **kwargs) → Any¶

init_indexes(ignore_duplicate_keys: bool = False) → None¶

as_any() → Any¶

get_current_request() → Request | None¶

get_endpoint_for_current_request() → Endpoint | None¶

add_background_task(func: Callable, *args, **kwargs) → None¶

Web Server¶

Declaring Endpoints¶

Registering Endpoints:¶

Resource REST Endpoints¶

Resource REST Endpoint with Parent¶

Validation¶

API References¶

Table of Contents

Related Topics

Useful Links