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.

  1. Using the endpoint decorator:

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

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

  2. 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
)

  3. 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.

  1. Manually using the WSGIApp.register_endpoint function:

    from superdesk.core.app import SuperdeskAsyncApp

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

  2. 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, 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)

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, 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, 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

endpoint(url: str, name: 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, cors: bool | None = None, tags: list[str] | None = None, summary: str | None = None, description: str | None = None, responses: dict | None = None)[source]

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

  • 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] | 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

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] | 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
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() → 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