`sanic-boom`¶

Components injection, fast routing and non-global (layered) middlewares. Give your Sanic application a Boom!

In a nutshell¶

"""Example code taken from
https://marshmallow.readthedocs.io/en/3.0/quickstart.html#quickstart
"""

import datetime as dt
import inspect
import typing as t

from marshmallow import Schema, fields, post_load
from sanic.exceptions import ServerError
from sanic.response import text

from sanic_boom import Component, SanicBoom

# --------------------------------------------------------------------------- #
# marshmallow related code
# --------------------------------------------------------------------------- #


class User(object):
    def __init__(self, name, email):
        self.name = name
        self.email = email
        self.created_at = dt.datetime.now()

    def __repr__(self):
        return "<User(name={self.name!r})>".format(self=self)

    def say_hi(self):
        return "hi, my name is {}".format(self.name)


class UserSchema(Schema):
    name = fields.Str()
    email = fields.Email()
    created_at = fields.DateTime()

    @post_load
    def make_user(self, data):
        return User(**data)


# --------------------------------------------------------------------------- #
# sanic-boom related code
# --------------------------------------------------------------------------- #


class JSONBody(t.Generic[t.T_co]):
    pass


class JSONBodyComponent(Component):
    def resolve(self, param: inspect.Parameter) -> bool:
        if hasattr(param.annotation, "__origin__"):
            return param.annotation.__origin__ == JSONBody
        return False

    async def get(self, request, param: inspect.Parameter) -> object:
        inferred_type = param.annotation.__args__[0]
        try:
            return inferred_type().load(request.json).data
        except Exception:
            raise ServerError(
                "Couldn't convert JSON body to {!s}".format(inferred_type)
            )


app = SanicBoom(__name__)
app.add_component(JSONBodyComponent)


@app.post("/")
async def handler(user: JSONBody[UserSchema]):  # notice the handler parameters
    return text(user.say_hi())


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, workers=1)

$ curl -v http://localhost:8000/ -d '{"name":"John Doe","email":"john.doe@example.tld"}'
*   Trying ::1...
* TCP_NODELAY set
* connect to ::1 port 8000 failed: Connection refused
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to localhost (127.0.0.1) port 8000 (#0)
> POST / HTTP/1.1
> Host: localhost:8000
> User-Agent: curl/7.61.1
> Accept: */*
> Content-Length: 50
> Content-Type: application/x-www-form-urlencoded
>
* upload completely sent off: 50 out of 50 bytes
< HTTP/1.1 200 OK
< Connection: keep-alive
< Keep-Alive: 5
< Content-Length: 23
< Content-Type: text/plain; charset=utf-8
<
* Connection #0 to host localhost left intact
hi, my name is John Doe

Warning

IMPORTANT: sanic-boom is in very early stages of development! Use with caution and be aware that some functionalities and APIs may change between versions until they’re out of alpha.

Dependencies¶

sanic-boom depends on two “not-so-known” libraries (both created by the author of sanic-boom):

sanic-ipware; and
xrtr

Important

Since xrtr replaces the Sanic default router under the hood in sanic-boom, it is very important for the developer to read its documentation (in the link provided above).

Documentation¶

https://sanic-boom.readthedocs.io/en/latest/

License¶

sanic-boom is a free software distributed under the MIT license.

Installing¶

To install sanic-boom, use pip (or your favorite Python package manager), like:

pip install sanic-boom

And you’re ready to the next step. Hooray!

Warning

At this time, sanic-boom is only available for Python 3.5+ (and probably will be).

Using `sanic-boom`¶

TBD.

Configuration¶

TBD.

`sanic-boom` inner works¶

TBD.

Layered middlewares¶

TBD.

Components¶

TBD.

Component lifecycle¶

TBD.

Component resolver¶

TBD.

Caching¶

TBD.

Routing¶

TBD.

Parameter parsing¶

TBD.

Headers¶

A simple example on how to pass the current request.headers as a component with sanic-boom.

TBD.

Marshmallow¶

This example shows how to integrate marshmallow with sanic-boom, in a way where you can have a generic type that, followed by a defined schema inside some route, will parse the response.json object given to that schema and return the created object to the handler.

Think of this as a “barebone simpler version” of DRF or any other serializer / deserializer.

"""Example code taken from
https://marshmallow.readthedocs.io/en/3.0/quickstart.html#quickstart
"""

import datetime as dt
import inspect
import typing as t

from marshmallow import Schema, fields, post_load
from sanic.exceptions import ServerError
from sanic.response import text

from sanic_boom import Component, SanicBoom

# --------------------------------------------------------------------------- #
# marshmallow related code
# --------------------------------------------------------------------------- #


class User(object):
    def __init__(self, name, email):
        self.name = name
        self.email = email
        self.created_at = dt.datetime.now()

    def __repr__(self):
        return "<User(name={self.name!r})>".format(self=self)

    def say_hi(self):
        return "hi, my name is {}".format(self.name)


class UserSchema(Schema):
    name = fields.Str()
    email = fields.Email()
    created_at = fields.DateTime()

    @post_load
    def make_user(self, data):
        return User(**data)


# --------------------------------------------------------------------------- #
# sanic-boom related code
# --------------------------------------------------------------------------- #


class JSONBody(t.Generic[t.T_co]):
    pass


class JSONBodyComponent(Component):
    def resolve(self, param: inspect.Parameter) -> bool:
        if hasattr(param.annotation, "__origin__"):
            return param.annotation.__origin__ == JSONBody
        return False

    async def get(self, request, param: inspect.Parameter) -> object:
        inferred_type = param.annotation.__args__[0]
        try:
            return inferred_type().load(request.json).data
        except Exception:
            raise ServerError(
                "Couldn't convert JSON body to {!s}".format(inferred_type)
            )


app = SanicBoom(__name__)
app.add_component(JSONBodyComponent)


@app.post("/")
async def handler(user: JSONBody[UserSchema]):  # notice the handler parameters
    return text(user.say_hi())


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, workers=1)


# then, run:
# curl -v http://localhost:8000/ -d '{"name":"John Doe","email":"john.doe@example.tld"}'

Tip

This example is just illustrative. You can use any other ODM-like library in here. Take a look on middle, another package by the same author of sanic-boom.

sanic-jwt¶

This example shows how to integrate sanic-jwt with sanic-boom, using a layered middleware, to determine that from a specific route onwards, every user will have to be authenticated - so you won’t need to declare anything else in your endpoints to protect them.

It also figures the usage of a component that is solely required by another component, as a simple example.

from sanic.response import text
from sanic_jwt import Initialize, exceptions

from sanic_boom import Component, ComponentCache, SanicBoom

# --------------------------------------------------------------------------- #
# sanic-jwt related code
# --------------------------------------------------------------------------- #


class User(object):
    def __init__(self, id, username, password):
        self.user_id = id
        self.username = username
        self.password = password

    def __str__(self):
        return "User(id='{}')".format(self.id)

    def to_dict(self):
        return {"user_id": self.user_id, "username": self.username}


users = [
    User(1, "user1", "abcxyz"),
    User(2, "user2", "abcxyz"),
    User(3, "user3", "abcxyz"),
    User(4, "user4", "abcxyz"),
]

username_table = {u.username: u for u in users}
userid_table = {u.user_id: u for u in users}


async def authenticate(request, *args, **kwargs):
    username = request.json.get("username", None)
    password = request.json.get("password", None)

    if not username or not password:
        raise exceptions.AuthenticationFailed("Missing username or password.")

    user = username_table.get(username, None)
    if user is None:
        raise exceptions.AuthenticationFailed("User not found.")

    if password != user.password:
        raise exceptions.AuthenticationFailed("Password is incorrect.")

    return user


# --------------------------------------------------------------------------- #
# sanic-boom related code
# --------------------------------------------------------------------------- #


class AuthComponent(Component):  # for shorthand
    def resolve(self, param) -> bool:
        return param.name == "auth"

    async def get(self, request, param):
        return request.app.auth


class JWTComponent(Component):
    def resolve(self, param) -> bool:
        return param.name == "jwt_user_id"

    async def get(self, request, param, auth):  # component inter-dependency
        is_valid, status, reasons = auth._check_authentication(
            request, None, None
        )
        if not is_valid:
            raise exceptions.Unauthorized(reasons, status_code=status)
        return auth.extract_user_id(request)

    def get_cache_lifecycle(self):
        return ComponentCache.REQUEST


app = SanicBoom(__name__)
sanicjwt = Initialize(app, authenticate=authenticate)

# adding components
app.add_component(AuthComponent)
app.add_component(JWTComponent)


@app.middleware(uri="/restricted")
async def restricted_middleware(jwt_user_id):
    pass  # this is really it!


@app.get("/restricted/foo")
async def restricted_handler(jwt_user_id):
    return text(jwt_user_id)


@app.get("/restricted/bar")
async def another_restricted_handler():
    return text("this is restricted!")


@app.get("/")
async def unrestricted_handler():
    return text("OK")


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, workers=1)

Testing¶

Calling the root endpoint:

$ curl -v http://127.0.0.1:8000/
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to 127.0.0.1 (127.0.0.1) port 8000 (#0)
> GET / HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/7.61.1
> Accept: */*
>
< HTTP/1.1 200 OK
< Connection: keep-alive
< Keep-Alive: 5
< Content-Length: 2
< Content-Type: text/plain; charset=utf-8
<
* Connection #0 to host 127.0.0.1 left intact
OK

So far so good, this was the expected result. Now, let’s try to access a restricted endpoint (by the code, any endpoint starting with /restricted/ will have authentication required), without a token:

$ curl -v http://127.0.0.1:8000/restricted/foo
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to 127.0.0.1 (127.0.0.1) port 8000 (#0)
> GET /restricted/foo HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/7.61.1
> Accept: */*
>
< HTTP/1.1 401 Unauthorized
< Connection: keep-alive
< Keep-Alive: 5
< Content-Length: 76
< Content-Type: application/json
<
* Connection #0 to host 127.0.0.1 left intact
{"reasons":["Authorization header not present."],"exception":"Unauthorized"}

But, but … Is that black magic? Actually, no. This is really straightforward. Now, let’s finally authenticate a user:

$ curl -v http://127.0.0.1:8000/auth -d '{"username":"user1","password":"abcxyz"}'
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to 127.0.0.1 (127.0.0.1) port 8000 (#0)
> POST /auth HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/7.61.1
> Accept: */*
> Content-Length: 40
> Content-Type: application/x-www-form-urlencoded
>
* upload completely sent off: 40 out of 40 bytes
< HTTP/1.1 200 OK
< Connection: keep-alive
< Keep-Alive: 5
< Content-Length: 140
< Content-Type: application/json
<
* Connection #0 to host 127.0.0.1 left intact
{"access_token":"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE1Mzk4OTUxODh9.FF2zld_RM01nhkFLVPIa6SRg6PZkGCCW6rFjrpTkc0o"}

Great, we have an access_token! Let’s try to access our restricted endpoint again:

$ curl -v http://127.0.0.1:8000/restricted/foo -H "Authorization: Bearer <token>"
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to 127.0.0.1 (127.0.0.1) port 8000 (#0)
> GET /restricted/foo HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/7.61.1
> Accept: */*
> Authorization: Bearer <token>
>
< HTTP/1.1 200 OK
< Connection: keep-alive
< Keep-Alive: 5
< Content-Length: 1
< Content-Type: text/plain; charset=utf-8
<
* Connection #0 to host 127.0.0.1 left intact
1

And our return is 1, as is the user_id parameter for user1. You can try to get a token for each user (user2, user3 and user4) and execute this last endpoint. The result should be the number of the user.

And what about the layered middleware? You just need to implement one argument in a middleware and all endpoints starting with it will run it, and in this example, will require an authenticated user. Another example? Sure!

$ curl -v http://127.0.0.1:8000/restricted/bar -H "Authorization: Bearer <token>"
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to 127.0.0.1 (127.0.0.1) port 8000 (#0)
> GET /restricted/bar HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/7.61.1
> Accept: */*
> Authorization: Bearer <token>
>
< HTTP/1.1 200 OK
< Connection: keep-alive
< Keep-Alive: 5
< Content-Length: 19
< Content-Type: text/plain; charset=utf-8
<
* Connection #0 to host 127.0.0.1 left intact
this is restricted!

Well, a lot of boilerplate code has just vanished ¯\_(ツ)_/¯

SQLAlchemy¶

TBD.

Reference¶

sanic-boom¶

class sanic_boom.BoomRequest(url_bytes, headers, version, method, transport)[source]¶

remote_addr¶

Attempt to return the original client ip based on X-Forwarded-For.

Returns:	original client ip.

class sanic_boom.ComponentCache[source]¶: An enumeration.

Contributing¶

Contributions are welcome, and they are greatly appreciated! Every little bit helps, and credit will always be given.

Bug reports¶

When reporting a bug please include:

Your operating system name and version.

Any details about your local setup that might be helpful in troubleshooting.

Detailed steps to reproduce the bug.

Documentation improvements¶

sanic-boom could always use more documentation, whether as part of the official sanic-boom docs, in docstrings, or even on the web in blog posts, articles, and such.

Feature requests and feedback¶

The best way to send feedback is to file an issue at https://github.com/vltr/sanic-boom/issues.

If you are proposing a feature:

Explain in detail how it would work.
Keep the scope as narrow as possible, to make it easier to implement.
Remember that this is a volunteer-driven project, and that code contributions are welcome :)

Development¶

To set up sanic-boom for local development:

Fork sanic-boom (look for the “Fork” button).

Clone your fork locally:

git clone git@github.com:your_name_here/sanic-boom.git

Create a branch for local development:
```
git checkout -b name-of-your-bugfix-or-feature
```
Now you can make your changes locally.
When you’re done making changes, run all the checks, doc builder and spell checker with tox one command:
```
tox
```

Commit your changes and push your branch to GitHub:

git add .
git commit -m "Your detailed description of your changes."
git push origin name-of-your-bugfix-or-feature

Submit a pull request through the GitHub website.

Pull Request Guidelines¶

If you need some code review or feedback while you’re developing the code just make the pull request.

For merging, you should:

Include passing tests (run tox) [1].
Update documentation when there’s new API, functionality etc.
Add a note to CHANGELOG.rst about the changes.
Add yourself to AUTHORS.rst.

[1]	If you don’t have all the necessary python versions available locally you can rely on Travis - it will run the tests for each change you add in the pull request. It will be slower though …

Tips¶

To run a subset of tests:

tox -e envname -- pytest -k test_myfeature

To run all the test environments in parallel (you need to pip install detox):

detox

Changelog¶

v0.1.2 on 2018-10-23¶

Added components property on BoomRequest so any request “leftover” may be handled properly (like an open database connection).

v0.1.1 on 2018-10-18¶

Fixed a bug where handlers derived from HTTPMethodView class were not being executed (for their signature actually be *args, **kwargs).

v0.1.0 on 2018-10-17¶

First release on PyPI. (Probably) not stable.

Authors¶

Richard Kuesters - https://vltr.github.io/

Thanks!¶

I need to thank everyone who gave an idea here and there regarding the scope of sanic-boom, specially to @ahopkins for some great ideas for caching and lifecycle 😎 🍻

sanic-boom¶

In a nutshell¶

Dependencies¶

Documentation¶

License¶

Installing¶

Using sanic-boom¶

Configuration¶

sanic-boom inner works¶

Layered middlewares¶

Components¶

Component lifecycle¶

Component resolver¶

Caching¶

Routing¶

Parameter parsing¶

Headers¶

Marshmallow¶

sanic-jwt¶

Testing¶

SQLAlchemy¶

Reference¶

sanic-boom¶

Contributing¶

Bug reports¶

Documentation improvements¶

Feature requests and feedback¶

Development¶

Pull Request Guidelines¶

Tips¶

Changelog¶

v0.1.2 on 2018-10-23¶

v0.1.1 on 2018-10-18¶

v0.1.0 on 2018-10-17¶

Authors¶

Thanks!¶

Inspirations¶

`sanic-boom`¶

Using `sanic-boom`¶

`sanic-boom` inner works¶