Multiple Clients

Sometime, projects requires multiple api that require different configuration, such as different middleware for authentication or discovery mechanism.

The pyramid-blacksmith plugin enabled that by name your client factory.

the default name is client, but it can be override using the blacksmith.clients setting.

This settings is a list such as

blacksmith.clients =
   client_foo
   client_bar

In that case, the request.blacksmith property will contains to functions, name client_foo and client_bar but there is no client.

Example

ini file

###
# app configuration
# http://docs.pylonsproject.org/projects/pyramid/en/latest/narr/environment.html
###

[app:main]
use = egg:notif

pyramid.includes =
    pyramid_blacksmith

blacksmith.scan =
    notif.resources

blacksmith.clients =
    client_static
    client_consul
    client_router

blacksmith.client_static.service_discovery = static
blacksmith.client_static.static_sd_config =
    user/v1       http://user:8000/v1
blacksmith.client_static.error_parser = notif:MyError

blacksmith.client_consul.service_discovery = consul
blacksmith.client_consul.consul_sd_config =
    addr                            http://consul:8500/v1
    service_name_fmt                {service}-{version}
    service_url_fmt                 http://{address}:{port}/{version}
    unversioned_service_name_fmt    {service}
    unversioned_service_url_fmt     http://{address}:{port}
blacksmith.client_consul.error_parser = notif:MyError


blacksmith.client_router.service_discovery = router
blacksmith.client_router.router_sd_config =
    service_url_fmt                 http://router/{service}-{version}/{version}
    unversioned_service_url_fmt     http://router/{service}
blacksmith.client_router.error_parser = notif:MyError


###
# wsgi server configuration
###

[server:main]
use = egg:waitress#main
host = 0.0.0.0
port = 8000

###
# Pserve configuration
###
[pserve]
watch_files =
    *
    %(here)s/notif/**/*.py
    /home/notif/.cache/pypoetry/virtualenvs/**/*.py

###
# logging configuration
# http://docs.pylonsproject.org/projects/pyramid/en/latest/narr/logging.html
###

[loggers]
keys = root, notif

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = INFO
handlers = console

[logger_notif]
level = DEBUG
handlers =
qualname = notif

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(asctime)s %(levelname)-5.5s [%(name)s][%(threadName)s] %(message)s

pyramid views

import email as emaillib
import smtplib
from textwrap import dedent

from blacksmith import SyncConsulDiscovery
from notif.resources.user import User
from pyramid.config import Configurator

smtp_sd = SyncConsulDiscovery()


def send_email(user: User, message: str):
    email_content = dedent(
        f"""\
        Subject: notification
        From: notification@localhost
        To: "{user.firstname} {user.lastname}" <{user.email}>

        {message}
        """
    )
    msg = emaillib.message_from_string(email_content)

    srv = smtp_sd.resolve("smtp", None)
    # XXX Synchronous socket here, OK for the example
    # real code should use aiosmtplib
    s = smtplib.SMTP(srv.address, int(srv.port))
    s.send_message(msg)
    s.quit()


def post_notif_using_static(request):
    if request.method == "GET":
        return {"detail": "Use POST to test the static driver"}

    body = request.json
    api_user = request.blacksmith.client_static("api_user")
    user: User = (api_user.users.get({"username": body["username"]})).unwrap()
    send_email(user, body["message"])
    return {"detail": f"{user.email} accepted"}


def post_notif_using_consul(request):
    if request.method == "GET":
        return {"detail": "Use POST to test the consul driver"}

    body = request.json
    api_user = request.blacksmith.client_consul("api_user")
    user: User = (api_user.users.get({"username": body["username"]})).unwrap()
    send_email(user, body["message"])
    return {"detail": f"{user.email} accepted"}


def post_notif_using_router(request):
    if request.method == "GET":
        return {"detail": "Use POST to test the router driver"}

    body = request.json
    api_user = request.blacksmith.client_router("api_user")
    user: User = (api_user.users.get({"username": body["username"]})).unwrap()
    send_email(user, body["message"])
    return {"detail": f"{user.email} accepted"}


def main(global_config, **settings):
    """Build the pyramid WSGI App."""
    with Configurator(settings=settings) as config:

        config.add_route("notify_v1", "/v1/notification")
        config.add_view(
            post_notif_using_static, route_name="notify_v1", renderer="json"
        )

        config.add_route("notify_v2", "/v2/notification")
        config.add_view(
            post_notif_using_consul, route_name="notify_v2", renderer="json"
        )

        config.add_route("notify_v3", "/v3/notification")
        config.add_view(
            post_notif_using_router, route_name="notify_v3", renderer="json"
        )

        app = config.make_wsgi_app()
        return app