Skip to main content

Microservice with django

Project description


bgtasks is python library for dealing with data exchange between micro services using rabbitmq protocol. Moreover, you can use it as trigger for events which belongs to another services.


Use the package manager pip to install bgtasks.

pip install bgtasks

AMQP = {
    'USERNAME': 'guest',
    'PASSWORD': 'guest',
    'VHOST': '/',
    'HOST': 'localhost',
    'PORT': 5672,
    'RPC_SLEEP_TIME': 0.005,
    'RPC_TIMEOUT': 5,

Default values



Creating bgtask view for handle actions through route


from bgtasks import rpc_tasks
from bgtasks import Response

def handle(data):
    return Response('I received your message %s' % data)

To get response

from bgtasks import RPCClient

rpc_client = RPCClient()

    response ='message', 'Hi')
except TimeoutError:
    print('Service is not responding')

In order to avoid conflicts between remote procedure calls you should pass parameters explicitly with keywords

To run rpc task run command below

python tasks




from django.db import models

class Category(models.Model):
    name = models.CharField(max_length=255)

In this case your add should receive arguments' list with explicit variable name ids


from bgtasks import rpc_tasks
from bgtasks import Response
from bgtasks import serializer_class
from testapp.models import Category
from rest_framework import serializers

class CategorySerializer(serializers.ModelSerializer):
    class Meta:
        model = Category
        fields = ('id', 'name')

@serializer_class(CategorySerializer, many=True)
def handle(serializer):
    return Response(

def handle(serializer):
    queryset = Category.objects.filter(id__in=serializer.validated_data['ids'])
    serializer = CategorySerializer(queryset, many=True)
    return Response(



from django.db import models
from bgtasks.models import RemoteField

class Product(models.Model):
    name = models.CharField(max_length=255)
    category = RemoteField() # in our case, it is in another service id


from rest_framework import serializers
from bgtasks.rest_framework.fields import RemoteField
from app1.models import Product

class ProductSerializer(serializers.ModelSerializer):
    category = RemoteField(route='get')

    class Meta:
        model = Product
        fields = ('id', 'name', 'category', )


And make sure that returned response should be formed as below format.

    'status': 'success/fail',
    'data': [
            'id': 1,
            # 'data'
            'id': 2,
            # data

Handling list serializer

In order to avoid from sending many rpc requests at first level of serializer we added RPCSerializerMixin

from bgtasks import RemoteField
from bgtasks.rest_framework.serilaizers import RPCSerializerMixin
from rest_framework import serializers
from app1.models import Product

class ProductListSerializer(RPCSerializerMixin, serializers.ModelSerializer):
    category = RemoteField(route='get')

    class Meta:
        model = Product
        fields = '__all__'

users = Product.objects.all()
serializer = ProductListSerializer(users, many=True)

It will send to route one request with gathered pks in body as [1,2,3,4,5], after which will be iterated to merge current serializer data which maps to id field in rpc response

        'id': 1,
        'name': 'IPhone',
        'category': {
            'id': 5,
            'name': 'Phone',
        'id': 2,
        'name': 'LG Smart Tv',
        'category': {
            'id': 3,
            'name': 'TV',

Merge methods

To handle many=True in serializer we introduce RPCSerializerMixin which uses merge functions. You can import them as below, and to understand can look to function profile.

from bgtasks.utils.merge import merge, merge_dict, merge_obj


Add ENVIRONMENT = 'test' on in order to imitate response from another service

import json
from django.test import TestCase
from django.test import Client
from bgtasks import rpc_tasks
from bgtasks import RPCClient
from bgtasks import SUCCESS
from bgtasks.amqp import register_tasks

def add_user(data):
    return 1

class RPCTestCase(TestCase):
    def setUp(self):
        register_tasks() # If you want to run your tasks to test them out, not only rpc tasks which are registered inside of your test file

    def test_add_user(self):
        data = {'username': 'john', 'password': 'smith'}
        c = Client()
        response ='/user/add/', data)
        data = json.loads(response.content)
        self.assertEqual(response.status_code, 201)
        self.assertEqual(data['user_id'], 1)

    def test_your_tasks(self):
        data = RPCClient().call('mytasks', {})
        self.assertEqual(data['status'], SUCCESS)


Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.

Please make sure to update tests as appropriate.

How to deploy

Create config file in home directory ~/.pypirc

repository = 
username = myrubapa

After run command for build and deploy

python3 sdist bdist_wheel
python3 -m twine upload dist/*

for more detail read packaging-projects



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

bgtasks-0.0.44.tar.gz (15.7 kB view hashes)

Uploaded source

Built Distribution

bgtasks-0.0.44-py3-none-any.whl (22.5 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