Servey - A Flexible Action Framework For Python

This project specifying metadata for python functions (In a manner similar to FastAPI) which is then used to build REST, GraphQL, and Scheduled services. These are locally runnable / runnable in a hosted environment using Starlette, Strawberry and Celery. They may also be run on AWS infrastructure using Serverless and Lambda. Tests and examples may also be specified for actions. General design goals are:

• We want to cover the basic utility that almost any application will require as simply as possible.
• Convention over configuration
• Configurability
• Openness and playing nicely with the other children - not just doing something, but exposing details of what is being done to external services
• We want the utility offered by things like AWS while being as minimally tied to them as possible.

Example

Install servey in your project using:

pip install servey[server]

Create a file actions.py containing the following:

from servey.action.action import action
from servey.trigger.web_trigger import WEB_GET


@action(triggers=(WEB_GET,))
def say_hello(name: str) -> str:
    """ Greet a user! """
    return f"Hello {name}!"

Note

• The action decorator indicates that the say_hello function will be special!
• The actions module (and any submodules of it) is the default location in which Servey will look for actions. This may be overridden by specifying a different value in the SERVEY_ACTION_PATH environment variable
• We specify a trigger for this action - WEB_GET
• Servey uses marshy to marshall arbitrary python objects.
• Servey uses schemey for schema generation / validation.

Run an action from the terminal

Actions should have unique names, which taken from the function name by default, but can also be overridden in the decorator. This name can be used to run an action explicitly from the command line (or cron):

python -m servey --run=action --action=say_hello "--event={\"name\": \"World\"}"

Run Server

Start the Starlette server using:

python -m servey

You should see console output regarding keys and temporary passwords (More on this in the Authorization section), as well as information indicating that Uvicorn is running on port 8000. (You override this using the SERVER_PORT environment variable)

The following endpoints deployed by default:

• http://localhost:8000/docs : OpenAPI Docs for your project
• http://localhost:8000/graphiql/ : GraphQL debugger for your project
• http://localhost:8000/openapi.json : OpenAPI Schema for your project
• http://localhost:8000/graphql : GraphQL endpoint for your project

Servey populates the OpenAPI Schema using the annotations on your function, the action decorator, and any documentation you provided.

Specifying Example Usage for Actions

You can specify action usage examples using the action decorator. These will be available in the OpenAPI schema as well as potentially being used to generate unit tests. Update your actions.py with the following:

from servey.action.action import action
from servey.action.example import Example
from servey.trigger.web_trigger import WEB_GET


@action(
    triggers=(WEB_GET,),
    examples=(
        Example(
            name='greet_developer',
            description='Say hello to the developer',
            params={'name': 'Developer'},
            result='Hello Developer!'
        ),
    )
)
def say_hello(name: str) -> str:
    """ Greet a user! """
    return f"Hello {name}!"

Restart the server, to update your OpenAPI schema.

Run pip install pytest, add an empty tests/__init__.py and then specify the following tests/test_actions.py:

from servey.servey_test.test_servey_actions import define_test_class

TestActions = define_test_class()

• Run tests with python -m unittest discover -s tests
• TestActions will include tests of all your examples from your actions where include_in_tests is True
• Nothing prevents you from creating your own unit tests for actions - they're just functions with an additional servey_action attribute!

Caching

Actions should be able to provide recommended caching strategies to clients. (The clients can ignore this of course!) Http caching available for REST endpoints, but not GraphQL - though technologies like React Query could be used to add it. Consider the following actions.py:

from datetime import datetime
from time import sleep

from servey.action.action import action
from servey.cache_control.ttl_cache_control import TtlCacheControl
from servey.trigger.web_trigger import WEB_POST


@action(triggers=(WEB_POST,), cache_control=TtlCacheControl(30))
def slow_get_with_ttl() -> datetime:
    """
    This function demonstrates http caching with a slow function. The function will take 3 seconds to return, but
    the client should cache the results for 30 seconds.
    """
    sleep(3)
    return datetime.now()

Notice that when you restart the server and run this from the OpenAPI test page, the first time it runs it should take ~3 seconds. Subsequent runs are instant as the disk cache retains the result for 30 seconds.

Authorization

Servey Provides a pluggable authorization mechanism. By default, Servey uses JWT tokens and scopes for authorization, with a key for generating them either specified in the JWT_SECRET_KEY environment variable or regenerated on each server restart. (The AWS Lambda implementation uses KMS by default for key storage.) Note that we are talking about Authorization here rather than Authentication.

Servey does not want to specify how a valid token should be issued, though we do include debug authenticator implementation based on OAuth2. It generates a random password which is printed to the logs on server restart. Alternatively you may specify a password in the SERVEY_DEBUG_AUTHENTICATOR_PASSWORD environment variable. A REAL Authenticator would be backed by a database of some kind, and could be plugged in to replace this one, or even run from a different server.

Actions may specify an access_control to limit access. Consider the following actions.py:

from typing import Optional

from servey.action.action import action
from servey.security.access_control.scope_access_control import ScopeAccessControl
from servey.security.authorization import Authorization
from servey.trigger.web_trigger import WEB_GET


@action(triggers=(WEB_GET,))
def echo_authorization(
    authorization: Optional[Authorization],
) -> Optional[Authorization]:
    """
    By default, authorization is derived from signed http headers - this just serves as a way
    of returning this info
    """
    return authorization


@action(triggers=(WEB_GET,), access_control=ScopeAccessControl(execute_scope='root'))
def only_for_root() -> str:
    """
    This can only be executed if the user has the root scope
    """
    return 'Some Secret Data!'

• The OpenAPI docs page now includes an OAuth2 section.
• echo_authorization gets the authorization from the http headers, decodes and confirms it and echos it. By default, we look for parameters with type Authorization and inject them from the context rather than directly from input parameters.
• only_for_root can only be executed by users with the root scope
• GraphQL uses the same access_controllers, reading tokens from the Authorization http header. (Graphiql lets you specify this)

Scheduler

So far we have demonstrated usage of WebTrigger, but triggers are pluggable and other implementations are possible. One additional type included is the FixedRateTrigger This allows you to specify that a function should run at regular intervals.

• In a single server / development environment, Background Threads are used.
• If the environment specifies a CELERY_BROKER, Servey uses Celery to run background tasks in a distributed fashion
• In a Serverless / AWS lambda environment, servey implements scheduling by deploying triggers for the generated lambdas

Here is a celery deployment example.

Nested Actions

Out of the box, actions may be defined on a function of a returned type, allowing for nested actions to be defined and resolved lazily in graphql. (Nested actions may have a WebTrigger too if required):

from dataclasses import dataclass

from servey.action.action import action
from servey.trigger.web_trigger import WEB_GET


@dataclass
class NumberStats:
    value: int

    @action
    async def factorial(self) -> int:
        """
        This demonstrates a resolvable field, lazily resolved (Usually by graphql)
        """
        result = 1
        index = self.value
        while index > 1:
            result *= index
            index -= 1
        return result


@action(
    triggers=(WEB_GET,),
)
def number_stats(value: int) -> NumberStats:
    return NumberStats(value)

• We define a return type NumberStats that is simply a python dataclass
• The field factorial is only resolved if requested in the graphql request
• Nested Actions may specify caching and access controls

Event Channels

Event Channels (Reworked from Subscriptions) model the case where events are sent somewhere outside the responsibility of the server - be it an external process, queue, webhook, or event to a connected client. Although pluggable (Like everything else in servey), there are 3 default implementations:

• BackgroundActionChannel Run an action as a background process (Using either asyncio, celery, or SQS depenending on circumstances) with the event as a parameter
• WebhookActionChannel Invoke a webhook with the event marshalled as a payload
• WebsocketEventChannel Send the event to connected clients over a websocket. An access controller determines who may connect to a channel, and an event filter hides events from users who should not get them. In an AWS environment, APIGateway, Lambda and Dynamodb are used to implement this. In a single server environment, Starlette is used. In a clustered server environment, Celery and Starlette are used.

Create a event_channels.py file with the following content:

from servey.event_channel.websocket.websocket_event_channel import (
    websocket_event_channel,
)

messenger = websocket_event_channel("messenger", str)

Open your actions.py and add the following:

from servey.action.action import action
from subscriptions import messager

# noinspection PyUnusedLocal
@action(triggers=(WEB_POST,))
def broadcast_message(message: str, connection_id: Optional[str] = None) -> bool:
    """ Send a message to all connected users or to a single subscriber. """
    messenger.publish(message, connection_id)
    return True

Restart the server, to go to https://localhost:8000/asyncapi.json

Unfortunately there is no studio where you can try it out with asyncapi like there is with OpenApi right now. I have been using the "Browser WebSocket Client" chrome extension to test subscriptions Using the url: ws://localhost:8000/subscription/messager/some_unique_subscriber_id) and the openapi docs to send messages.

You might have noticed that we do not actually implement graphql subscriptions. The reason for this is we wanted to provide a unified interface for subscriptions across all platforms, and the way appsync implements Graphql subscriptions is quite frankly, weird. (Each subscription is triggered by a mutation, there is no admin interface, you trigger the subscription by invoking the graphql mutation. Even if you can secure these, you end up with mutations which are not useful to most users. And don't get me started on event filtering

type TriggerMessageEvent { subscriber_id: string event: Message }

Mutation { triggerMessage(subscriber_id: string, event: Message): TriggerMessageEvent }

Subscription { message(subscriber_id: string): TriggerMessageEvent }

AWS

Up until this point, we have mostly discussed development environments / deploying to a container. Servey also allows your code to be deployed to AWS using Serverless. Servey will generate serverless definitions in yaml files in order to facilitate this. We assume that you already have an aws account with appropriate access, and that you are set up with serverless (You probably have a $HOME/.aws/credentials file set up). First, you'll need some extras to get this working:

pip install servey[serverless]

Then you can regenerate your serverless.yml definitions using:

python -m servey --run=sls

• This will generate a new serverless.yml file for you if it is missing. (override environment variable MAIN_SERVERLESS_YML_FILE to choose a different name)
• Servey uses file includes to attempt to make the modifications to the main serverless yaml minimal.
• Actions get implemented as Lambdas - one per action.
• We implemented GraphQL using Appsync
• We implemented REST using API Gateway
• We implemented Authorizers using KMS
• We implements subscriptions to actions using SQS
• The generated lambdas as designed to allow direct invocation where the event contains unmarshalled parameters, or access by Appsync or API Gateway.
• Once you deploy your serverless project, you should be able to test from the Appsync, Api Gateway, and Lambda consoles respectively.

Templating

Although the focus of the project is on building out REST / Graphql APIs, we also included an integration with the Jinja2 Templating Engine, and the ability to deploy static files. Actions are linked to templates by means of a WebPageTrigger

@action(triggers=(WebPageTrigger(),))
def current_time_page() -> datetime:
    """
    No template or path were defined, so these are derived from the action name
    ('/current-time-page' and 'templates/current_time_page.j2' respectively
    """
    return datetime.now() 

The template is passed the result of the action as a model variable. e.g.:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Current Time</title>
</head>
<body>
<h1>The current time is {{ model }}</h1>
</body>
</html>

This allows opportunities for bootstrapping.

Note: We currently do not automatically deploy static files to AWS - it is assumed you will add S3 / Cloudfront / Route53 resources to your serverless definition manually, as there is a lot of variability in how you may want to set this up. We do however include an example that includes S3, Route53 and cloudfront (here)[examples/b_end_2_end]

Dynamic / Generated Actions

There are cases where actions must be dynamically generated, and servey supports this. Consider the following example from an actions.py, which will create 3 dynamically generated actions:

def _generate(action_name: str):
    @action(name=action_name, triggers=WEB_GET)
    def my_action() -> str:
        return f"action_name was {action_name}"
    return my_action


generated_1 = _generate('generated_1')
generated_2 = _generate('generated_2')
generated_3 = _generate('generated_3')

These are handles in the AWS Lambda environment slightly differently from regular actions - they share a lambda rather than each action having their own. (Though you can specify this behaviour for regular actions too by setting the SERVEY_AWS_ROUTER_FOR_ALL to 1 when running python -m servey --run=sls)

servey_router:
  handler: servey.servey_aws.lambda_router.invoke
  timeout: 900
  events:
  - http:
      path: /actions/generated-1
      method: get
      cors: true
  - http:
      path: /actions/generated-2
      method: get
      cors: true
  - http:
      path: /actions/generated-3
      method: get
      cors: true

An example deployment for this is located at examples/c_generated

Command line tools

Produce an openapi schema in openapi.json:

python -m servey --run=openapi

Produce a graphql schema in servey_schema.graphql:

python -m servey --run=graphql-schema

Pluggability

We use marshy for pluggable components. See (marshy_config_servey)[marshy_config_servey/init.py]

Deployment Patterns

• API in ApiGateway / AppSync, SPA hosted on S3 and cloudfront out in front, Deployment of all via serverless.
• Docker Image containing Nginx / Starlette app deployed to Heroku / Linode.

Installing local development dependencies

python setup.py install easy_install "servey[dev]"

Release Procedure

The typical process here is:

• Create a PR with changes. Merge these to main (The Quality workflows make sure that your PR meets the styling, linting, and code coverage standards).
• New releases created in github are automatically uploaded to pypi