DEV Community: Pavel Vergeev

Django Celery Beat: How to get the last time a PeriodicTask was run

Pavel Vergeev — Thu, 13 Feb 2020 21:28:39 +0000

The problem

Suppose you have the following task:

import time

from django_project.celery import app  # set up according to https://docs.celeryproject.org/en/stable/django/first-steps-with-django.html#using-celery-with-django


@app.task
def my_task():
    print("I'm running!")
    time.sleep(10)
    print("I've finished!")

Suppose further my_task runs once in several days using django-celery-beat in a single worker process.

The question is: how can my_task get the last time it was run?

Take number one: Bind and Get

The first thing that comes to mind is, find where django-celery-beat puts the last running time of a PariodicTask and take that value.

So first you need to bind the task in order to get its name, then get the appropriate PeriodicTask and take its last_run_at attribute, like so:

@app.task(bind=True)
def my_task(self):
    last_run = PeriodicTask.objects.get(
        task=self.name  # notice PeriodicTask.name and self.name are different things
    ).last_run_at
    print(f"I'm running! Last time I ran at {last_run_at}")
    time.sleep(10)
    print("I've finished!")

The problem: PeriodicTasks will accumulate each time you restart celery process. Soon you'll be getting

django_celery_beat.models.PeriodicTask.MultipleObjectsReturned: get() returned more than one PeriodicTask -- it returned 16!

Take number two: Bind and Filter

In order to avoid MultipleObjectsReturned exception, we can filter out all irrelevant periodic tasks:

@app.task(bind=True)
def my_task(self):
    last_run = PeriodicTask.objects.filter(
        task=self.name
    ).order_by('last_run_at').last().last_run_at
    print(f"I'm running! Last time I ran at {last_run_at}")
    time.sleep(10)
    print("I've finished!")

The first problem: If we run my_task more often than once in five minutes, last_run_at is going to get duplicated:

[2020-02-13 20:15:41,884: WARNING/ForkPoolWorker-2] I'm running! Last time I ran at 2020-02-13 17:13:00.616557+00:00
[2020-02-13 20:15:51,886: WARNING/ForkPoolWorker-2] I've finished!
[2020-02-13 20:15:51,956: INFO/ForkPoolWorker-2] Task integration.tasks.test_task[1e413b12-455c-4740-8438-d6d1f8b18a35] succeeded in 10.409016128978692s: None
[2020-02-13 20:16:01,787: WARNING/ForkPoolWorker-2] I'm running! Last time I ran at 2020-02-13 17:13:00.616557+00:00
[2020-02-13 20:16:11,788: WARNING/ForkPoolWorker-2] I've finished!
[2020-02-13 20:16:11,825: INFO/ForkPoolWorker-2] Task integration.tasks.test_task[80a4467f-d0b2-4839-a0a4-1270bf02c2e8] succeeded in 10.18468949093949s: None
[2020-02-13 20:16:21,734: INFO/MainProcess] Received task: integration.tasks.test_task[5d4c126e-da9d-4986-8d37-d2d9a072bf60]
[2020-02-13 20:16:21,906: WARNING/ForkPoolWorker-2] I'm running! Last time I ran at 2020-02-13 17:16:01.635268+00:00
[2020-02-13 20:16:31,907: WARNING/ForkPoolWorker-2] I've finished!
[2020-02-13 20:16:31,939: INFO/ForkPoolWorker-2] Task integration.tasks.test_task[5d4c126e-da9d-4986-8d37-d2d9a072bf60] succeeded in 10.203773486078717s: None

Fortunately, the problem does not persist if the task is run once in several days. So it would be a valid solution, as long as you don't need to change the schedule of the periodic task.

The second problem: Suppose you do need to change the schedule of the periodic task. You actually cannot do that, or at least I haven't found a way to do that with code. So after PeriodicTask gets reinstantiated with the new schedule, last_run_at becomes None.

The third problem: last_run_at also becomes None if you want to change the time zone.

So what I'm saying is, last_run_at is not a reliable solution for a long-running task.

Take number three: Store it somewhere else

By this time I knew what I wanted: a reliable way to store the last running time of a task that gets executed rarely in a single process. So I've decided to just store it in the database:

# models.py

class PeriodicTaskRun(models.Model):
    task = models.CharField(max_length=200, verbose_name='Task Name')
    created_at = models.DateTime(auto_now_add=True)

# tasks.py

@app.task(bind=True)
def my_task(self):
    last_run = PeriodicTaskRun.objects.filter(task=self.name).latest()
    PeriodicTaskRun.objects.create(task=self.name)

    print(f"I'm running! Last time I ran at {last_run.created_at}")
    time.sleep(10)
    print("I've finished!")

If you know any caveats to this solution I should know about, or if you have a better one, please let me know in the comments!

Let's Run a Tor Bridge! Quick Reference Guide

Pavel Vergeev — Tue, 03 Sep 2019 16:45:23 +0000

In its recent newletter the Tor Project asked for help with setting up new, uncensored bridges. Since this is a good way to contribute to the project, I went ahead and did it. While doing so, I've compiled this quick reference guide on how to run a bridge relay and why doing so is a good idea.

What is Tor

Watch a video overview of Tor

TL;DR: Tor is a network for circumventing censorship and protecting identity online.

How Tor works

Read detailed description of Tor inner workings

TL;DR: The Tor browser connects to a network of publicly-listed interconnected nodes. They toss the packets to each other in a randomized fashion, anonymizing the sender of the packets.

What does a Tor bridge do

Read documentation for the Tor bridges

TL;DR: Bridge relays allow to connect to Tor network even if it's banned by the government. Also, the bridges hide the fact that a person uses Tor.

Why should you run a bridge

Read the post asking to run more bridges

TL;DR: It allows people living in oppressive regimes to have access to the open internet. Free access to information is a basic human right.

How to run a bridge

Rent a VPS or use the one already at your disposal. I found DigitalOcean the most convenient provider to use, but went ahead with my local provider to reduce the cost of running the relay.
Follow the installation instructions. Docker installation is the easiest one.

Monitoring

Post-install notes describe the process. You can also count the number of users by viewing the relay logs. It can take up to several month before your bridge gets its first user. It's advised to be patient and wait until your bridge accumulates users.

TL;DR: Go to the relay search, and type in your relay fingerprint. Here's the example of the page you'll get. Bookmark it, and you'll be able to monitor your bridge activity.

Big thanks to Philipp Winter of the Tor Project for helping me to set up my first bridge.

If you find yourself in need of advice, don't hesitate to email me to pavel@vergeev.site.

Productive Full-Stack Development with PyCharm

Pavel Vergeev — Sun, 27 Jan 2019 20:30:33 +0000

It took me a long time to figure out which area of software development I like. So I've tried a bunch of different technologies, and many different tools that accompany them. I've spent three years coding with Vim, wrote a whole project using Netbeans, and gradually tried every popular text editor for web development out there, including VSCode.

Seeing VSCode getting so much attention, with people like Kenneth Reitz recommending it for Python development, I thought it would be refreshing to talk about completely different development setup.

I discovered some PyCharm features after I saw them used in other editors. With this post, I hope to do the same for you: encourage you to explore your editor more thoroughly, getting the most out of it.

Recognizing the strengths

There is a couple of things IDEs do best, so I tried to center my workflow around them:

Navigating the codebase.
Integrating everything in one place.

What follows is the list of features I use almost every day. I also give links to appropriate documentation pages. If there's a keyboard shortcut, I will indicate it in brackets (having in mind Apple keyboard layout).

Navigating the codebase

Common-sense features

A common thing to have among the editors is "find everything". In PyCharm, it's ⇧⇧ shortcut.

"Jump to declaration" (⌘B) would not surprise anyone, too. In PyCharm it works in many different contexts, with no configuration whatsoever. So if I wanted to find SCSS class declaration, I would press ⌘B and get

Same shortcut for JS symbols, Django Template Language, and simply opening an included file. And if I use ⌘B on the declaration itself, it will show the usages.

To find a string, I use global search (⌘⇧F). The nice thing about it is that it allows me to search in the source code of the project dependencies, even if they are located in a different folder.

The "take me back" button (⌥⌘←) comes in handy when I found what I was looking for.

There's also multi-cursor with different editing shortcuts that I like, but I'll avoid reciting the manual. Everybody has these features anyway.

Things I didn't know I needed

Ubiquitous and reliable hints is something I always miss when I'm using a simple text editor.

I don't have to google the bootstrap classes anymore! It also hints me the names of the attributes I override, a thing I tend to do all the time while working with Django.

The hints are also useful for imports. But I don't use them for that, because PyCharm has built-in auto import.

Vim bindings is something I wanted from every editor I tried. This is not the case with PyCharm with its select-nearest-block-that-makes-sense feature (⌥↑):

Finally, the bonus for non-native English speakers such as myself: the spellcheck. In one of my projects, we had two classes: Enhancement and EnchancementTemplate. My teammate made a typo in the second name, and it would have remained unnoticed if PyCharm didn't underline Enchancement during the code review.

Integrating everything in one place

Deployment

I don't fully remember how to deploy the project I'm working on.

All I know is that clicking the green "start" button at the corner will do all the necessary work for me. It will even reload the server once I change the source code.

This is enabled by the thing PyCharm calls Configurations. In Django project, all this thing asks for is environment variables. But it allows me to do much more than that.

In my project, Django runs on its own, but Redis and Postgres run in their own containers. So I made my Django configuration run docker-compose up -d each time I start the project. This way, I lifted the mental toll of having to remember this detail.

Debugging

Once I configured how I run the project, I got the debugger configuration for free. The only difference is that I click on the green bug instead of the green triangle.

In order to insert a breakpoint, I don't need to insert any code. I click on the line I want my breakpoint to be on, and trigger the target code. No more forgetting to remove import pdb; pdb.set_trace() for me!

When I hit the breakpoint, I click "Evaluate Expression" (the little calculator icon) and here I have the same interactive shell I had in pdb with an addition of hints and autocomplete.

I love that debugger behaves nicely even if two or more requests come in and trigger the same breakpoint several times. With pdb, this situation was a nightmare.

VCS and task tracking

When I first started using PyCharm, I did all my version control from the terminal. To make things comfy, I used a lot of git aliases. Then I dug a little deeper into how PyCharm handles the VCS and liked it.

I don't need to remember the details of how VCS is handled. Any time I have something to commit, I press ⌘K, deselect the things I don't want to commit yet, and write the commit message. As someone who's obsessed with git add --patch and atomic commits, I find it wonderful that I can visually select the changes I'm going to commit.

Now that PyCharm knows when I commit, it would be nice if it could track the ticket I'm commiting for, right? And this can be done: I connected the corporate Gitlab server with Tools | Tasks & Contexts | Configure Servers and PyCharm fetched the tickets. Now I have different sets of tabs and changes open for each task, and can switch to a new task more easily. I can even measure the time I spent working in each context and send it to my issue tracker.

Sometimes I forget to commit often and end up losing important changes. Local History allows me to roll them back and already saved me hours of repetitive work.

Terminal

Two great things about it:

It remembers the opened tabs and history for each of them.
So one of my tabs is always opened in case I need SSH Port Forwarding, and in the other I do fab deploy to deploy the code to production.
Virtual Environment is activated automatically, so pip install works exactly as I expect.

The glaring problem with it is inability for the terminal to share environment with Configurations without copy-paste, so python manage.py makemigrations might end up doing migrations for the wrong database. I show how to ameliorate the problem in the next section.

Python Console

It's nice to be able to run arbitrary code inside the Django project. The problem is that I have to remember to set the correct environment for python manage.py shell. And again, I want my hints and autocomplete.

To achieve that, I install a couple of external things:

EnvFile plugin for PyCharm
dotenv package for Python

I tell my configurations to load variables from the .env file, like so:

Then I go Preferences | Build, Execution, Deployment | Console | Django Console and in the Starting script area write

import sys; print('Python %s on %s' % (sys.version, sys.platform))
import django; print('Django %s' % django.get_version())
sys.path.extend([WORKING_DIR_AND_PYTHON_PATHS])

# this will allow to load the correct environment for the shell
from dotenv import load_dotenv
from pathlib import Path
load_dotenv(dotenv_path='.env')

# this is needed if the project uses django-configurations package
# import configurations
# configurations.setup()

if 'setup' in dir(django): django.setup()
import django_manage_shell; django_manage_shell.run(PROJECT_ROOT)

# and this will allow to call any "python manage.py ..." command from the shell
from django.core.management import call_command

Now I can open Python Console to run migrations and freely interact with Django models while getting autocomplete and the default IDE shortcuts.

In this very shell I can even interact with the data on the remote database. To do that, I forward the remote database to my local host (see how remote port forwarding works):

ssh -L 5431:127.0.0.1:5432 user@example.org

And then I tell Django that the production database is located at 127.0.0.1:5432 using Django's support for multiple databases.

Now the following code in the Python Console will get me the number of users on the production database:

User.objects.using('production').count()

Jupyter Notebook

Integration of Jupyter Notebook with Django is similar to integrating Python Console. Open the .ipynb in PyCharm. As the first cell, run

import sys; print('Python %s on %s' % (sys.version, sys.platform))
import django; print('Django %s' % django.get_version())
sys.path.extend([WORKING_DIR_AND_PYTHON_PATHS])

# this will allow to load the correct environment for the shell
from dotenv import load_dotenv
from pathlib import Path
load_dotenv(dotenv_path='.env')

# this is needed if the project uses django-configurations package
# import configurations
# configurations.setup()

django.setup()

and you have the Django-Jupyter-PyCharm toolchain ready to go.

Using port forwarding trick described above, you can even do the computations on the production data.

Database

If you didn't like the idea of local code messing with production data, you can connect to the database using a dedicated tool.

It was of great help when the ORM was not available. And again, autocomplete and database inspection made life so much easier.

This thing is also well-documented.

Sass

In order to get Sass/SCSS files compile automatically, I had to set up file watchers.

For my project, the default settings were OK, but generally file watchers ask you to provide the path where you want to put the compiled files.

Scratches

I'm writing this post in PyCharm. I use a scratch file, and the process looks like this:

Alongside the scratch file, I have the example project opened.

I use scratch files primarily while I'm designing some complex features.

Scratch files are also easily exported to Github Gist:

Updating PyCharm

Until recently, I used to update PyCharm by downloading a new version manually from the website and replacing the binary.

Now there's a tool for that.

Ready to explore?

I came up with the workflow I currently have because other people shared how they do stuff. I got familiar with the idea of the one-step build thanks to Makefile. I found integrated debugger in PyCharm after I read about that exact feature for VSCode. I decided I needed comfortable Python shell when I saw my friend using IPython.

This is why I wrote this post: so you could discover new features you like and integrate them into your workflow.

I don't want to stop with what I have. Let me know how you prefer to get things done, I probably have a lot to learn.

Also

The Law of Leaky Abstractions — why you shouldn't use an IDE if you only starting out.
Vim Is The Perfect IDE — while I am avoiding CLI as much as possible, this post offers the opposite approach.
Curse of the IDE — problems with IDEs you should be aware of.
Master your IDE — a short post with direct instructions.

Testing against unmanaged models in Django

Pavel Vergeev — Tue, 15 Aug 2017 19:58:12 +0000

The problem

My Django application is supposed to read the tables of an already existing database. So the models of the database are generated using inspectdb command. They look like this:

class User(models.Model):
    first_name = models.CharField(max_length=255)
    second_name = models.CharField(max_length=255)
    # other properties

    class Meta:
        managed = False
        db_table = 'user'

The managed = False property tells Django not to create the table as part of migrate command and not to delete it after flush. The catch here is that Django won't create the table during testing either. So when I wanted to test my code against non-production database, I ran into the following issues:

The need to create all of the tables for unmanaged models by hand just to run the tests. Otherwise I was getting django.db.utils.OperationalError: no such table: user.
It was not obvious to me how to generate fixtures for the unmanaged models. If the model is managed, you can run the migrations on the test database, populate the tables with the help of shell command and save it with dumpdata. But the tables of unmanaged models are not created during migrations, there's nothing to populate. Again, I had to create everything by hand.

My goal was to find a way to automate the process, making both testing and collaboration easier. Relevant bits and pieces of are scattered across the Internet. This is an attempt to collect them in one place.

The simple solution

Before the test runner launches the tests, it runs the migrations creating the tables for managed models. The migrations themselves are Python code that creates the models and sets the managed property. It is possible to modify a migration so that it sets managed to True while we are testing and False otherwise for unmanaged models.
To do that, we will create IS_TESTING variable in settings.py and set it to True when we are testing. We will also modify the migration code:

from django.conf import settings

# ...
operations = [
    migrations.CreateModel(
        name='User',
        fields=[
            # ...
        ],
        options={
            'db_table': 'user',
            # 'managed': False,
            'managed': settings.IS_TESTING,
        },
    ),
]

Now the table will be created whenever the migration is run with IS_TESTING = True.
The idea belongs to Kyle Valade, who described it in his blog.
To generate fixtures, the method with shell command described earlier will work.
The downside here is that you have to remember to modify the migration of every unmanaged model.

Creating a custom test runner

A more complex solution is to create a custom test runner that will convert all unmanaged models to managed before running a test and revert the effect afterwards.
We'll put the runner in appname/utils.py:

from django.test.runner import DiscoverRunner


class UnManagedModelTestRunner(DiscoverRunner):

    def setup_test_environment(self, *args, **kwargs):
        from django.apps import apps
        get_models = apps.get_models
        self.unmanaged_models = [m for m in get_models() if not m._meta.managed]
        for m in self.unmanaged_models:
            m._meta.managed = True
        super(UnManagedModelTestRunner, self).setup_test_environment(*args, **kwargs)

    def teardown_test_environment(self, *args, **kwargs):
        super(UnManagedModelTestRunner, self).teardown_test_environment(*args, **kwargs)
        for m in self.unmanaged_models:
            m._meta.managed = False

Now we will tell Django to use this runner by adding TEST_RUNNER = 'app_name.utils.UnManagedModelTestRunner' to our settings.
We are not yet ready because the User model has custom table name user. This is why we need to create the test tables without running migrations. There's a small app for that. It's installed by running pip install and adding it to INSTALLED_APPS. Our tests will work if we run them with -n switch: python manage.py test -n. As a consequence, we will lose the ability to see if any of our migrations are broken (which is probably fine if all of them are generated by Django).

The idea of creating a custom test runner belongs to Tobias McNulty, who posted it in Caktus Group blog. The code from his post had to be updated.

When it comes to maintaining the code, there are complications. First, if we wanted to use some other test runner, we'd have to inherit from it:

from django.test.runner import DiscoverRunner
from django_nose import NoseTestSuiteRunner


class UnManagedModelTestRunner(NoseTestSuiteRunner, DiscoverRunner):
    # ...

Second, even if the django-test-without-migrations app is simple, it doesn't mean it can't be broken by a new version of Django, so we need to be prepared to troubleshoot it.
Third, we have to generate fixtures in an unusual way. The tables of our unmanaged models are only available in setUp() method, so to generate the fixtures we would have to add and dump data in the source code of the test:

import sys

from django.core.management import call_command

# ...

    def setUp(self):
        models.User.objects.create(
            # ...
        )
        sysout = sys.stdout
        sys.stdout = open('fixtures/users.json', 'w')
        call_command('dumpdata', app_label='app_name')
        sys.stdout = sysout

After we ran the code, we can remove it and load the fixture in our tests the usual way.

The common weakness

When the tests run, they will treat the unmanaged models as managed. They won't fail if someone accidentally adds a field to an unmanaged model. The only way I know to get around this is to create the tables by hand.

Final word

This is all I've got on the topic. If you happen to know another solution, I'd be happy to learn about it.

How to test input processing in Python 3

Pavel Vergeev — Tue, 25 Jul 2017 17:02:03 +0000

Testing functions that use input() is not a straightforward task. This tutorial shows how to do that.

The problem

Contest programming problem solutions are usually tested automatically. So the input is just a set of strings with predefined format that fed into standard input.

Suppose, for example, a problem requires us to read in N stacks of different sizes. Then input might look like this:

Here, 3 is the number of stacks to read in. The first number on each subsequent line is the size of the stack (which is not required in Python, but is helpful in other languages), followed by its elements.

To extract data from the input, a few manipulations have to be made:

def get_input_stacks():
    n = int(input())
    stacks = []
    for _ in range(n):
        str_stack = input().split(' ')
        stack = [int(s) for s in str_stack]
        stacks.append(stack)
    return stacks

And it doesn't take a lot of effort to make a mistake here. In fact, I already made one in the code above. So, it would be nice to have automatic tests for this kind of functions too.

The solution

One of the solutions I found on StackOverflow when applied to my problem looks like this:

from unittest.mock import patch
import unittest

import containers

class ContainersTestCase(unittest.TestCase):

    def test_get_input_stacks_processed_input_correctly(self):
        user_input = [
            '3',
            '4 1 2 3 2',
            '1 2',
            '0',
        ]
        expected_stacks = [
            [1, 2, 3, 2],
            [2],
            [],
        ]
        with patch('builtins.input', side_effect=user_input):
            stacks = containers.get_input_stacks()
        self.assertEqual(stacks, expected_stacks)


if __name__ == '__main__':
    unittest.main()

When I run the test, I see that it fails:

The problem is that I read in the size of the stack as an element of the stack. Instead, I should just ignore it: str_stack = input().split(' ')[1:].

How the solution works

This is the most interesting line in the test:

with patch('builtins.input', side_effect=user_input):

I assume it is known how with statement works. unittest.mock.patch has several keyword arguments that modify its behavior, but in this case, it simply replaces built-in input() function with a unittest.mock.MagicMock object. One of the properties of the object is to return its return_value unless its side_effect returns something else:

>>> import unittest.mock
>>> mock = unittest.mock.MagicMock()
>>> mock.return_value = '43'
>>> mock()
'43'
>>> def foo():
...  return 'something_else'
... 
>>> mock.side_effect = foo
>>> mock()
'something_else'

Notice that side_effect, unlike return_value, is a function. side_effect can also be an exception object (then the exception will be raised when the mock is called) or an iterable. If it's an iterable, the mock yields a new value every time it's called:

>>> mock.side_effect = [1, 2]
>>> mock()
1
>>> mock()
2
>>> mock()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/local/Cellar/python3/3.6.1/Frameworks/Python.framework/Versions/3.6/lib/python3.6/unittest/mock.py", line 939, in __call__
    return _mock_self._mock_call(*args, **kwargs)
  File "/usr/local/Cellar/python3/3.6.1/Frameworks/Python.framework/Versions/3.6/lib/python3.6/unittest/mock.py", line 998, in _mock_call
    result = next(effect)
StopIteration
>>>

and this is what we are trying to achieve by mocking input(): a different predefined value each time it is called. This is why we are passing side_effect=user_input to the patch() function (side_effect is both a property of MagicMock and a keyword argument of patch()):

user_input = [
    '3',
    '4 1 2 3 2',
    '1 2',
    '0',
]
with patch('builtins.input', side_effect=user_input):

The argument 'builtins.input' contains a name that points to the input() function. What patch() does is it basically makes this name point to the MagicMock object until the end of the with block. An object in Python can have multiple names, so patch the right one (here's how). builtins is a module where built-in function names are located:

>>> import builtins
>>> builtins.print('hi')
hi

Each time the function get_input_stacks() calls the patched input() it gets the next string in the user_input list. This is the behavior we wanted.

Where to go from here

To get a better understanding of the capabilities offered by unittest.mock, it is best to read the manual. As an exercise, I would suggest testing a function that does the reverse of get_input_stacks() by patching print().

Thank you for reading!