Skip to content

Instantly share code, notes, and snippets.

@Quard
Quard / bechmark_results.md
Created Jul 16, 2019
Benchmark results for article "My story about Go, Python and benkmarks."
View bechmark_results.md

Python + Redis

API get

Running 30s test @ http://rpi:5000/api/v1/url/aaaa
  12 threads and 400 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   489.29ms  196.29ms   1.38s    82.02%
 Req/Sec 68.43 31.64 192.00 72.98%
@Quard
Quard / run_queues_example.sh
Created Apr 10, 2019
The Python Celery Cookbook: Small Tool, Big Possibilities
View run_queues_example.sh
celery -A myproject worker -l info -Q celery
celery -A myproject worker -l info -Q mail
@Quard
Quard / queues_config_example.py
Created Apr 10, 2019
The Python Celery Cookbook: Small Tool, Big Possibilities
View queues_config_example.py
CELERY_TASK_ROUTES = {
'myproject.apps.mail.tasks.send_mail_task': {'queue': 'mail', },
}
@Quard
Quard / primary_celery_example_use.py
Created Apr 10, 2019
The Python Celery Cookbook: Small Tool, Big Possibilities
View primary_celery_example_use.py
send_mail_task.delay(('noreply@example.com', ), 'Celery cookbook test', 'test', {})
@Quard
Quard / primary_celery_example.py
Last active Dec 15, 2021
The Python Celery Cookbook: Small Tool, Big Possibilities
View primary_celery_example.py
from django.conf import settings
from django.core.mail import send_mail
from django.template import Engine, Context
from myproject.celery import app
def render_template(template, context):
engine = Engine.get_default()
@Quard
Quard / postponed_task_example.py
Created Apr 10, 2019
The Python Celery Cookbook: Small Tool, Big Possibilities
View postponed_task_example.py
from datetime import datetime
send_mail_task.apply_async(
(('noreply@example.com', ), 'Celery cookbook test', 'test', {}),
countdown=15 * 60
)
send_mail_task.apply_async(
(('noreply@example.com', ), 'Celery cookbook test', 'test', {}),
@Quard
Quard / long_running_task_example.py
Last active Apr 12, 2019
The Python Celery Cookbook: Small Tool, Big Possibilities
View long_running_task_example.py
@celery_app.task
def send_good_morning_mail_task(offset=0, limit=100):
users = User.objects.filter(is_active=True).order_by('id')[offset:offset + limit]
for user in users:
send_good_morning_mail(user)
if len(users) >= limit:
send_good_morning_mail_task.delay(offset + limit, limit)
@Quard
Quard / get_task_result_in_view_example.py
Created Apr 10, 2019
The Python Celery Cookbook: Small Tool, Big Possibilities
View get_task_result_in_view_example.py
def view(request):
find_job = find_best_service_provider_for_user.delay(request.user.pk)
# do other stuff
calculations_results = find_job.get().join()
# process calculations_results and send response
@Quard
Quard / get_task_result_example.py
Created Apr 10, 2019
The Python Celery Cookbook: Small Tool, Big Possibilities
View get_task_result_example.py
from celery import group
@celery_app.task
def calculate_service_provider_task(user_id, provider_id):
user = User.objects.get(pk=user_id)
provider = ServiceProvider.objects.get(pk=provider_id)
return calculate_service_provider(user, provider)
@Quard
Quard / celery_retries_example.py
Created Apr 10, 2019
The Python Celery Cookbook: Small Tool, Big Possibilities
View celery_retries_example.py
@celery_app.task(bind=True, default_retry_delay=10 * 60)
def send_mail_task(self, recipients, subject, template, context):
message = render_template(f'{template}.txt', context)
html_message = render_template(f'{template}.html', context)
try:
send_mail(
subject=subject,
message=message,
from_email=settings.DEFAULT_FROM_EMAIL,
recipient_list=recipients,