v0.0.4 Plugins / Extensions

[cirospaciari]

What is changed in socketify:

• ASGI lifespan fixes
• WSGI spec fixes
• WSGI and ASGI Server header from C++
• Plugins / Extensions support added
• Custom JSON Serializer support added
• Decorator Router
• res.send and res.cork_send

Custom JSON Serializer

socketify by default uses built-in json module with has great performance on PyPy, but if you wanna use another module instead of the default you can just register using app.json_serializer(module)

from socketify import App
import ujson
app = App()

# set json serializer to ujson
# json serializer must have dumps and loads functions
app.json_serializer(ujson)

app.get("/", lambda res, req: res.end({"Hello":"World!"}))

Plugins / Extensions

You can add more functionality to request, response, and websocket objects, for this you can use app.register(extension) to register an extension.
Be aware that using extensions can have a performance impact and using it with request_response_factory_max_items, websocket_factory_max_items
or the equivalent on CLI --req-res-factory-maxitems, --ws-factory-maxitems will reduce this performance impact.

Extensions must follow the signature def extension(request, response, ws), request, response, and ws objects contain method decorator that binds a method to an instance,
and also a property(name: str, default_value: any = None) that dynamic adds an property to the instance.

from socketify import App, OpCode

app = App()

def extension(request, response, ws):
    @request.method
    async def get_user(self):
        token = self.get_header("token")
        return { "name": "Test" } if token else { "name", "Anonymous" }
    
    @response.method
    def msgpack(self, value: any):
        self.write_header(b'Content-Type', b'application/msgpack')
        data = msgpack.packb(value, default=encode_datetime, use_bin_type=True)
        return self.end(data)
    
    @ws.method
    def send_pm(self, to_username: str, message: str):
        user_data = self.get_user_data()
        pm_topic = f"pm-{to_username}+{user_data.username}"
        
        # if the topic exists just send the message
        if app.num_subscribers(pm_topic) > 0:
            # send private message
            return self.publish(pm_topic, message, OpCode.TEXT)
        
        # if the topic does not exist create it and signal the user
        # subscribe to the conversation
        self.subscribe(pm_topic)
        # signal user that you want to talk to and create a pm room
        # all users must subscribe to signal-{username}
        self.publish(f"signal-{to_username}", { 
            "type": "pm", 
            "username": user_data.username, 
            "message": message 
        }, OpCode.TEXT)
    # this property can be used on extension methods and/or middlewares
    request.property("cart", [])

# extensions must be registered before routes
app.register(extension)

Decorator Router

As the name suggests this router allows to use of decorators for routing and also comes up with a prefix option, and middleware support.

from socketify import App

app = App()
router = app.router()

@router.get("/")
def home(res, req):
   res.end("Hello World!")

api = app.router(prefix="/api")

# will serve in /api/hello
@api.get("/hello") 
def hello(res, req):
   res.end("Hello API!")

private = app.router("/api", auth_middleware)

# will serve in /api/users and use auth_middleware
@private.get("/users")
def get_users(res, req, auth):
   res.end("Hello private API!")

res.send and res.cork_send

res.send(message, content_type=b'text/plain, status=b'200 OK', headers=None, end_connection=False) and res.cork_send(message, content_type=b'text/plain', status=b'200 OK', headers=None, end_connection=False)
combines res.write_status(), res.write_headers(), and res.end() in a way that is easier to use, if you want to send all in one call just using named parameters. Headers can receive any iterator of iterators/tuple like iter(tuple(str, str)) where the first value is the header name and the following the value, using res.cork_send will make sure to send all the
data in a corked state.

def not_found(res, req):
    res.send("Not Found", status=404)

def ok(res, req):
    res.send("OK", status="200 OK")

def json(res, req):
    res.send({"Hello", "World!"})

def with_headers(res, req):
    res.send({"Hello": "World!"}, headers=(("X-Rate-Limit-Remaining", "10"), (b'Another-Headers', b'Value')))

Work in progress for v0.1.0:

• Native optimizations Add native optimizations #52
• on_start, on_shutdown hooks Add on_start and on_shutdown lifespan hooks #58
• Cache tools LRU Cache and Timed Cache avoiding GIL a in memory cache solution #60
• Hot reloading Add hot reloading on CLI Tools #61

---

This discussion was created from the release v0.0.4 Plugins / Extensions.