DEV Community: Andrei Koptev

How to store passwords and keys in your project

Andrei Koptev — Sat, 03 Apr 2021 10:20:46 +0000

Prepare

Django project ready to deploy
Many private keys:
- Database username/password
- Twilio username/password
- Django SECRET_KEY
- ... many others

Solution

Yes, you may to save variable directly in your code. But this is a bad practice. Because not secure save keys in Github or another git solution!

1.Use Environment variables

import os

os.environ['MY_VAR'] = 'Private value' # Set key-value
my_var = os.environ.get('MY_VAR') # Read variable value from anywhere

Cons: after restart server we loose Environment variables

2.Save Environment variables in .bashrc (.zshrc)

Add to your ~/.zshrc file:

...

export MY_VAR="Private value" # Here!

# source ~/.zshrc

Good. Now we can to read our variable:

import os

my_var = os.environ.get('MY_VAR') # return 'Private value'

Bonus

For Django use useful environ-wrapper: django-environ
Create .env file, past private data to it and this is all:

env = environ.Env()
environ.Env.read_env()

SECRET_KEY = env('SECRET_KEY')

# Database
# https://docs.djangoproject.com/en/2.1/ref/settings/#databases
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql_psycopg2',
        'NAME': env('DB_NAME'),
        'USER': env('DB_USER'),
        'PASSWORD': env('DB_PASSWORD'),
        'HOST': 'localhost',
        'PORT': '',
    }
}

Don't forget to add .env to your .gitignore

Asynchronous log with API and PUSH. Part 1

Andrei Koptev — Thu, 01 Apr 2021 11:00:49 +0000

One of solution asynchronous log. With Rest api and optional PUSH notification to ios/android devices

Our application consist of 2 parts:

Notifier listen changes from models and fill notification model. With Rest API
Pusher listen changes from Notifier (Notification model) and send push to ios/android device

Requirements

Notification model must be available through API (list of notifications for concrete user)
Notification model have optional case: PUSH. If yes - send PUSH (ios/android)
Our application must be separate from other applications (is like as service)
Use only Django signals to fill notification model
asynchronous
Different icons, title, body and data in each message
Notifier work only in one direction: listen changes from models -> create notification in model -> (optional) send push

Important

Full code of application here
How to install and configure celery here

Part 1: Notifier

For example, we want to create notification and send push in case:

New article was created (Article model)
New comment was created (ArticleComment model)
User A send gift to user B (Gift model)

Ok. Let's get a started!

Firstly, create notifier application inside Django project
Create models.py inside it and add:

class Notification(models.Model):
    uid = models.UUIDField(default=uuid.uuid4,
                           editable=True,
                           unique=True)
    name = models.CharField(max_length=250)
    body = models.CharField(max_length=250)
    recipients = models.ManyToManyField(get_user_model())
    is_notify_screen = models.BooleanField(default=True)
    is_push = models.BooleanField(default=True)
    client_action = models.CharField(choices=ClientAction.choices,
                                     default=ClientAction.NONE,
                                     max_length=250)

    icon = models.CharField(blank=True,
                            max_length=250,
                            choices=Icon.choices,
                            default=Icon.INFO)

3.Register models.
Now we want to register models and connect it with post_save signal.

Our application must be fully separate and work only in one direction: listen changes and create notification.

Inside your app create tasks.py and add:

models = [
    Article,
    ArticleComment,
    Gift,
]

def register_models():
    for model in models:
        post_save.connect(_post_save,
                           sender=model,
                           dispatch_uid=uuid.uuid4())


def _post_save(sender, instance, **kwargs):
    """
    Handler for our post_save.
    PostSaveContext is a namedtuple - context from sync post_save to async post_save handler

    """
    created = kwargs.get( 'created' )
    if not created:
        return

    class_name = instance.__class__.__name__
    action = f'{class_name}_POST_SAVE'.upper()

    post_save_context = PostSaveContext( module=instance.__module__,
                                         class_name=class_name,
                                         instance_identifier=instance.pk,
                                         action=action )

    transaction.on_commit(
        lambda: _async_post_save_handler.delay( post_save_context._asdict() )
    )


@app.task
def _async_post_save_handler(post_save_context):
    perform_action( post_save_context )


def perform_action(post_save_context):
    """
    The core of notifier

    """
    module = post_save_context.get( 'module' )
    class_name = post_save_context.get( 'class_name' )
    instance_identifier = post_save_context.get( 'instance_identifier' )
    action = post_save_context.get( 'action' )

    module = module.split( '.' )[1]
    model = apps.get_model( app_label=module,
                            model_name=class_name )

    instance = model.objects \
        .filter( pk=instance_identifier ) \
        .first()

    if instance is None:
        return

Good!.
Now we may listen creation instances from each model in models and transfer information to _async_post_save_handler.

The next step:fill our notification model

Because we have different content from models, I think the good solution create context fabric (ContextFactory) and pass instance and action to it.

4.Create factory.py file and past to it:

class ContextFactory:
    def __init__(self, instance, action,  *args, **kwargs):
        self.instance = instance
        self.action = action
        self.__dict__.update(kwargs)

    def create_context(self):
        if self.action == NotifierAction.ARTICLE_POST_SAVE.value:
            recipients = self.instance.recipients.all()
            body = self.instance.body
            data = Data(name='New article available',
                        body=body,
                        recipients=recipients,
                        client_action=ClientAction.OPEN_ARTICLE,
                        is_notify_screen=True,
                        is_push=True,
                        icon=Icon.MAIL)

            return data

        if self.action == NotifierAction.ARTICLE_COMMENT_POST_SAVE.value:
            recipients = self.instance.article.recipients.all()
            data = Data(name='New comment',
                        body=self.instance.body,
                        client_action=ClientAction.OPEN_MESSAGE,
                        recipients=recipients,
                        is_notify_screen=True,
                        is_push=True,
                        icon=Icon.INFO)
            return data

        if self.action == NotifierAction.GIFT_POST_SAVE.value:
            recipients = [self.instance.owner]
            data = Data(
                name='Gift for you!',
                body='You got a new gift!',
                recipients=recipients,
                client_action=ClientAction.OPEN_GIFTS,
                is_notify_screen=True,
                is_push=True,
                icon=Icon.GIFT
            )

            return data

Ok. Now we may create notification instance. I like to use Managers for models.
Yep, many developers like to use services to create instance or change it.
But for me Managers is a good place to any model manipulations.

5.Create manager file and past to it:

class NotificationManager(models.Manager):
    def create_notification(self, context: Data):
        context_dict = context._asdict()
        instance = self.create(**context_dict)  # Create notification firstly
        if instance:
            for recipient in context.recipients:
                instance.recipients.add(recipient)  # Add User to m2m

class Notification(models.Model):
    ....
    actions = NotificationManager()

6.Now please add in perform_action method:

def perform_action(post_save_context):
     ...

    factory = ContextFactory(instance, action)
    context = factory.create_context()  # Create context to our model
    Notification.actions.create_notification(context)  # Create new notification

7.Finally, we must to start models listener:

from django.apps import AppConfig

class NotifierConfig(AppConfig):
    name = 'apps.notifier'

    def ready(self):
        from apps.notifier.tasks import register_models
        register_models() # Here!

Run celery, run project and create instance Dialog or Gift.
Now in our Notification model we can see new notifications.

8.Create api python package inside notifier. And add:

serializers.py
views.py
urls.py

views.py for our module:

class NotificationListGenericApiView(generics.ListAPIView):
    """
    Get a list of notifications for a current user with pagination. Readonly

    """
    serializer_class = NotificationSerializer
    queryset = Notification.objects.all()

    def get_queryset(self):
        return Notification\
            .actions\
            .for_user(self.request.user)

...and urls:

urlpatterns = [
    path('', NotificationListGenericApiView.as_view())
]

Now we can to list all notifications for a requested user.

Summary

Our notifier as a separate module and only listen post_save signal from models
Our notifier as scalable. Add new model and add context and it is all
Our notifier has own API
notifier work asynchronous

In the next part I show you pusher - listen changes from notifier and send push notification to mobile devices

How to install and configure celery

Andrei Koptev — Mon, 29 Mar 2021 08:53:32 +0000

Early in my articles I wrote about use celery, but not wrote how to install and configure it.

Prepare

You have Django project
You want to connect project with celery

Short description

Install broker (redis) globally
Install celery python library
Configure celery in settings
Write async function
Run celery

Detail

Install Redis:

sudo apt install redis-server

Check:

➜  ~ redis-cli
127.0.0.1:6379> ping
PONG
127.0.0.1:6379>

Install celery:

pip install celery

Create config file. Please don't name at celery because may be names conflicts. For example, run_celery.py:

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from django.conf import settings

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'settings')
app = Celery('my_celery_application')
app.config_from_object('django.conf:settings')

settings.py

BROKER_URL = 'redis://127.0.0.1:6379/0'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

Check all:

(ienv) ➜  project git:(master) ✗ celery -A run_celery worker -l info

 -------------- celery@MacBook-Pro-User1.local v5.0.2 (singularity)
--- ***** ----- 
-- ******* ---- macOS-10.16-x86_64-i386-64bit 2021-02-14 13:26:14
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         ptt:0x7fe9f133c2e0
- ** ---------- .> transport:   redis://127.0.0.1:6379/0
- ** ---------- .> results:     redis://127.0.0.1:6379/
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery


[tasks]
  . apps.centrifuge.centrifuge.send_sms_async

Write you async task:

from run_celery import app
from django.conf import settings
from .models import PhoneCode

@app.task
def send_sms_async(identifier: int):
    code = PhoneCode.objects.filter(pk=identifier).first()
    if code:
        provider: SMSProviderBase = Test()
        provider.send_private_sms(code.phone, code.code)

Conclusion

Celery with redis (or another backend) is a good way for your asynchronous Django projects
You may create as many tasks as you wish
You may choose a class-way for celery tasks
To control use flower to control your asynchronous tasks
To run celery via supervisor (-B - to use periodic tasks):

[program:example-celery]
command = <path/to/your/project/ienv>/bin/celery --app=run_celery:app worker --loglevel=INFO -B
directory = <path/to/your/project/>
user = <user>
stdout_logfile = <path/to/your/project/ienv>logs/celery.log
redirect_stderr=true
environment = LANG=en_US.UTF-8,LC_ALL=en_US.UTF-8

Thanks for reading

Django rest framework writable nested serializer

Andrei Koptev — Fri, 26 Mar 2021 20:49:27 +0000

Prepare

You have Django and Django rest framework
You have User with Profile as one-to-one
You have serializers:

class UserProfileSerializer(serializers.ModelSerializer):
    class Meta:
        model = UserProfile
        fields = [
            'experience',
            'rate',
        ]

class UserSerializer(serializers.ModelSerializer):
    avatar = ImageBase64Field(required=False)
    profile = UserProfileSerializer(required=False)

    class Meta:
        model = User
        fields = [
            'uid',
            'first_name',
            'last_name',
            'profile'
        ]

Problems

You want to update experience field in user profile

Solution

Add update method in your UserSerializer class.

class UserSerializer(serializers.ModelSerializer):
    ...

    def update(self, instance, validated_data):
        if validated_data.get('profile'):
            profile_data = validated_data.get('profile')
            profile_serializer = UserProfileSerializer(data=profile_data)

            if profile_serializer.is_valid():
                profile = profile_serializer.update(instance=instance.profile,
                                                    validated_data=profile_serializer.validated_data)
                validated_data['profile'] = profile

        return super().update(instance, validated_data)

Now, with user data we may to send profile data and update our user profile.

Conclusion

Now we may to update user profile easily in one request

Thanks for reading

Making Django's Signals Async with Celery

Andrei Koptev — Fri, 26 Mar 2021 20:42:16 +0000

Prepare

Django project with long-task logic
Celery with any backend

For example: send push after any event or auth user through sms code or email link.

Mark: to sign-in/sign-up users via SMS you may use this library

Problems

Django signals synchronous by default
We must to wait when transaction completed
Clients can to wait response for a long time

Short description

Configure celery
Create application (for example, notifier)
Configure post_save signal through celery task
Configure AppConfig to load function

Detail

Create notifier application
Add tasks.py

tasks.py

import uuid
from collections import namedtuple
from django.db.models.signals import post_save
from run_celery import app

from some_apps.models import SomeModel1, SomeModel2, SomeModel3

Ctx = namedtuple('Ctx', 'class_name identifier')

models = [
  SomeModel1, 
  SomeModel2, 
  SomeModel3
]


def register_models():
    for model in models:
        post_save.connect(_post_save,
                          sender=model,
                          dispatch_uid=uuid.uuid4())


def _post_save(sender, instance, **kwargs):
    ctx = Ctx(class_name=instance.__class__.__name__,
                        identifier=instance.pk)
    _async_post_save_handler.delay(ctx)


@app.task
def _async_post_save_handler(ctx: Ctx):
    """
    Get model and instance from notifier object and then make
    some other logic
    """
  ...

Finally, add to apps.py:


from django.apps import AppConfig


class NotifierConfig(AppConfig):
    name = 'notifier'

    def ready(self):
        from apps.notifier.tasks import register_models
        register_models() # here!

Save and run Django and run celery.

Conclusion

celery help you create async handlers to your Django models
Solution is great to separate post_save and async logic

Thanks for reading

Django image resize on the fly

Andrei Koptev — Thu, 11 Feb 2021 13:14:20 +0000

Init

Django model with image field

Purpose

Resize image to some template size (typed size)

Short

Catch original image from field
Resize it through PIL and save to BytesIO output
Send output to ContentFile
Send ContentFile to File
Run self.avatar.save(<name>, file, save=False) in model save method

Detail

ResizeImageMixin

import uuid
from PIL import Image
from io import BytesIO
from django.core.files import File
from django.core.files.base import ContentFile
from django.db import models


class ResizeImageMixin:
    def resize(self, imageField: models.ImageField, size:tuple):
        im = Image.open(imageField)  # Catch original
        source_image = im.convert('RGB')
        source_image.thumbnail(size)  # Resize to size
        output = BytesIO()
        source_image.save(output, format='JPEG') # Save resize image to bytes
        output.seek(0)

        content_file = ContentFile(output.read())  # Read output and create ContentFile in memory
        file = File(content_file)

        random_name = f'{uuid.uuid4()}.jpeg'
        imageField.save(random_name, file, save=False)

models.py

class User(Base, AbstractUser, ResizeImageMixin):
    def __str__(self):
        return f'{self.username}'

    def save(self, *args, **kwargs):
        if self.pk is None:
            self.resize(self.avatar, (200, 200))

        super().save(*args, **kwargs)