Welcome to Happyly’s documentation!

Introduction

Happyly is a scalable solution for systems which handle any kind of messages.

Happyly helps to abstract your business logic from messaging stuff, so that your code is maintainable and ensures separation of concerns.

Have you ever seen a codebase where serialization, message queue managing and business logic are mixed together like a spaghetti? I have. Imagine switching between Google Pub/Sub and Django REST Framework. Or Celery. This shouldn’t be a nightmare but it often is.

Here’s the approach of Happyly:

  • Write you business logic in universal Handlers, which don’t care at all how you serialize things or send them over network etc.

  • Describe your schemas using ORM/Framework-agnostic technology.

  • Plug-in any details of messaging protocol, serialization and networking. Change them with different drop-in replacements at any time.

Happyly can be used with Flask, Celery, Django, Kafka or whatever technology which can be used for messaging and also provides first-class support of Google Pub/Sub.

Use cases

  • Google Pub/Sub

    Let’s be honest, the official Python client library is too low-level. You must serialize and deserialize things manually, as well as to ack and nack messages.

    Usual way:

    def callback(message):
    attributes = json.loads(message.data)
    try:
        result = process_things(attributes['ID'])
        encoded = json.dumps(result).encode('utf-8')
        PUBLISHER.publish(TOPIC, encoded)
    except NeedToRetry:
        _LOGGER.info('Not acknowledging, will retry later.')
    except Exception:
        _LOGGER.error('An error occured')
        message.ack()
    else:
        message.ack()

    Happyly way:

    def handle_my_stuff(message: dict):
    try:
        return process_things(message['ID'])
    except NeedToRetry as error:
        raise error from error
    except Exception:
        _LOGGER.error('An error occured')

    handle_my_stuff is now also usable with Celery or Flask. Or with yaml serialization. Or with message.attributes instead of message.data. Without any change.

  • You are going to change messaging technology later.

    Let’s say you are prototyping your project with Flask and are planning to move to Celery for better fault tolerance then. Or to Google Pub/Sub. You just haven’t decided yet.

    Easy! Here’s how Happyly can help.

    1. Define your message schemas.

    class MyInputSchema(happyly.Schema):
    request_id = marshmallow.fields.Str(required=True)

class MyOutputSchema(happyly.Schema):
    request_id = marshmallow.fields.Str(required=True)
    result = marshmallow.fields.Str(required=True)
    error = marshmallow.fields.Str()

    1. Define your handler

    def handle_things(message: dict):
    try:
        req_id = message['request_id']
        if req_id in ALLOWED:
            result = get_result_for_id(req_id)
        else:
            result = 'not allowed'
        return {
            'request_id': req_id
            'result': result
        }
    except Exception as error:
        return {
            'request_id': message['request_id']
            'result': 'error',
            'error': str(error)
        }

    1. Plug it into Flask:

    @app.route('/', methods=['POST'])
def root():
    executor = happyly.Executor(
        handler=handle_things,
        deserializer=DummyValidator(schema=MyInputSchema()),
        serializer=JsonifyForSchema(schema=MyOutputSchema()),
    )
    request_data = request.get_json()
    return executor.run_for_result(request_data)

    1. Painlessly switch to Celery when you need:

    @celery.task('hello')
def hello(message):
    result = happyly.Executor(
        handler=ProcessThings(),
        serializer=happyly.DummyValidator(schema=MyInputSchema()),
        deserializer=happyly.DummyValidator(schema=MyOutputSchema()),
    ).run_for_result(
        message
    )
    return result

    1. Or to Google Pub/Sub:

    happyly.Listener(
    handler=ProcessThings(),
    deserializer=happyly.google_pubsub.JSONDeserializerWithRequestIdRequired(
        schema=MyInputSchema()
    ),
    serializer=happyly.google_pubsub.BinaryJSONSerializer(
        schema=MyOutputSchema()
    ),
    publisher=happyly.google_pubsub.GooglePubSubPublisher(
        topic='my_topic',
        project='my_project',
    ),
 ).start_listening()

    5. Move to any other technology. Or swap serializer to another. Do whatever you need while your handler and schemas remain absolutely the same.

Installation

Happyly is hosted on PyPI, so you can use:

pip install happyly

There is an extra dependency which enables cached components via Redis. If you need it, install it like this:

pip install happyly[redis]

Concepts

Handler

Handler is the main concept of all Happyly library. Basically a handler is a callable which implements business logic, and nothing else:

  • No serialization/deserialiation here

  • No sending stuff over the network

  • No message queues’ related stuff

Let the handler do its job!

To create a handler you can simply define a function which takes a dict as an input and returns a dict:

def handle_my_stuff(message: dict):
    try
        db.update(message['user'], message['status'])
        return {
            'request_id': message['request_id'],
            'action': 'updated',
        }
    except Exception:
        return {
            'action': 'failed'
        }

Done! This handler can be plugged into your application: whether it uses Flask or Celery or whatever.

Note that you are allowed to return nothing if you don’t actually need a result from your handler. This handler is also valid:

def handle_another_stuff(message: dict):
    try
        neural_net.start_job(message['id'])
        _LOGGER.info('Job created')
    except Exception:
        _LOGGER.warning('Failed to create a job')

If you prefer class-based approach, Happyly can satisfy you too. Subclass happyly.Handler() and implement the following methods:

class MyHandler(happyly.Handler):

    def handle(message: dict)
        db.update(message['user'], message['status'])
        return {
            'request_id': message['request_id'],
            'action': 'updated',
        }

    def on_handling_failed(message: dict, error)
        return {
            'action': 'failed'
        }

Instance of MyHandler is equivalent to handle_my_stuff

Executor

To plug a handler into your application you will need happyly.Executor() (or one of its subclasses).

Advanced

Indices and tables