Skip to main content

Django services app

Project description


This package provides services for django applications. The servitin service is an asynchronous standalone application.


pip install servitin


Add servitin to settings.INSTALLED_APPS

Make sure you have LOGGING settings in your project - servitin needs this.

Let's say you have a django app myapp.

Make myapp a servitin service by adding the line is_servitin = True in myapp/

from django.apps import AppConfig

class MyappConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'myapp'
    is_servitin = True

Create file myapp/

from servitin.base import BaseService

class Service(BaseService):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)"Myapp service ready.")

(Not necessary) Give the version myapp, write in the file myapp/

__version__ = '1.0'

Start service:

./ run_servitin

If all is ok you will see in the log Myapp service ready.


To configure the built-in logger for each service, Servitin looks for a logger named servitin_myservice_logger for the myservice app, similarly for the my_other_service app it will look for servitin_my_other_service_logger.

Buil-in logger is used like this:"Myapp service ready.")

in the example above

Request handling

Edit myapp/ to make the service as the ZeroMQ server:

from servitin.base import BaseService
from servitin.lib.zmq.server import ZMQServer

class Service(ZMQServer, BaseService):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)"Myapp service ready.")

Create myapp/

from django.conf import settings

settings.SERVITIN_MYAPP_ZMQ = getattr(settings, 'SERVITIN_MYAPP_ZMQ', {
    'BIND_ADDRESS': 'tcp://*:5555',
    'CONNECT_ADDRESS': 'tcp://',
    'SECRET': ''

Create file myapp/

import asyncio
from servitin.lib.zmq.server import zmq, Response
from asgiref.sync import sync_to_async
from django.core import serializers
from django.contrib.auth.models import User
from servitin.utils import serializable

async def get_users(request):
    order_by =['order_by']
    # use built-in logger'request order: {order_by}', name='@get_users')

    def get_data():
        # data with datetime objects, so it no serializable
        data = serializers.serialize('python', User.objects.all().order_by(order_by), fields=('username', 'date_joined'))
        # so make it serializable (you can use your own serializer) 
        return serializable(data)
    return Response(request, await sync_to_async(get_data)())

async def heavy_task(request):
    """ emulate heavy task endpoint for timeout test """

    await asyncio.sleep(5)'data: {}', id=request.request_id, name='@heavy_task')
    return Response(request, f'complete: {}')

Here we have created two endpoints: get_users, heavy_task. The service is ready to handle requests, let's test it.

Create django management command myapp/management/commands/

import asyncio
from import BaseCommand
from servitin.lib.zmq.client import Servitin
import myapp.settings  # import our service settings

class Command(BaseCommand):
    def handle(self, *args, **options):

        async def do():
            async with Servitin('myapp').connect() as my_service:
                print(await my_service.get_users({'order_by': 'username'}))

In this example servitin client configured via myapp/

Run it:

./ test_myapp_service


It is also possible to use the servitin client outside of django.

Create file

from servitin.lib.zmq.client import Servitin
import asyncio

loop = asyncio.get_event_loop()
params = {'connect': 'tcp://', 'secret': ''}


async def do():
    # manual close connection
    my_service = Servitin(**params)
    results = await asyncio.gather(*[
        my_service.get_users({'order_by': 'username'}),
        my_service.get_users({'order_by': 'id'})
    my_service.close()  # close connection

    # auto close connection via context manager
    async with Servitin(**params).connect() as my_service:
        print(await my_service.get_users({'order_by': 'username'}))

    # timeout
    # you can specify timeout by passing it to Servitin():
    # async with Servitin(connect='tcp://', secret='', timeout=1).connect() as my_service:
    async with Servitin(**params).connect() as my_service:
            # or pass it directly in call
            print(await my_service.heavy_task({'data': 'context manager, timeout call'}, timeout=1))
        except asyncio.TimeoutError as e:



params = {'connect': 'tcp://', 'secret': '', 'async_mode': False}

# manual
my_service = Servitin(**params)
print(my_service.get_users({'order_by': 'username'}))

# via context manager
with Servitin(**params).sync_connect() as my_service:
    print(my_service.get_users({'order_by': 'username'}))


Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

servitin-1.0.4.tar.gz (13.0 kB view hashes)

Uploaded source

Built Distribution

servitin-1.0.4-py3-none-any.whl (14.3 kB view hashes)

Uploaded py3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page