Skip to content

Instantly share code, notes, and snippets.

View ask's full-sized avatar
🎯
Focusing

Ask Solem ask

🎯
Focusing
2k followers · 107 following
View GitHub Profile
@ask
ask / shipments.py
Last active January 3, 2025 02:51
from enum import Enum
import faust
from faust import cli
from faust import web
class State(Enum):
# State of a shipment.
ISSUED = 'ISSUED'
CREATED = 'CREATED'
@ask
ask / finally_in_asynciterator_does_not_work_in_Python3.7
Last active March 22, 2019 23:54
# finally: behavior changed from Python 3.6 to Python 3.7
async def foo():
try:
async for item in Bar():
raise Exception('foo')
except Exception:
pass
@ask
ask / async_cached_property.py
Last active February 28, 2018 11:57
async_cached_property
import asyncio
from typing import Any, Awaitable, Callable, Optional, Type
# eh, like the @cached_property idiom, but async and you do `await x.y`
class async_cached_property:
name: str
def __init__(self, getter: Callable[[], Awaitable]) -> None:
@ask
ask / keybase.md
Created March 8, 2016 01:26

Keybase proof

I hereby claim:

  • I am ask on github.
  • I am asksol (https://keybase.io/asksol) on keybase.
  • I have a public key ASDn1BRTWvEf99dvp86-UNeZveCEufqJLdULQyCX36ulZAo

To claim this, I am signing this object:

@ask
ask / async.py
Last active August 29, 2015 14:01 — forked from mikeywaites/async.py
#project.async
from celery import Celery
celery = Celery(autofinalize=False)
def configure_celery(flask_app):
#...set celery up
celery.config_from_object(app.config)
@ask
ask / a_rabbitmq_test.py
Last active December 10, 2015 18:48
#!/usr/bin/env python
import logging
from pika import BlockingConnection, ConnectionParameters
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO)
@ask
ask / gist:2014539
Created March 11, 2012 01:57
celery consumer service using boot-steps.
from celery import abstract
from celery import current_app
from kombu import Exchange, Queue
from kombu.mixins import ConsumerMixin
# need to subclass the result backend so that it uses a topic exchange
# instead of direct, and send the results for tasks using a routing_key
# of the format:
@ask
ask / gist:1834966
Created February 15, 2012 10:43 — forked from sleekslush/gist:1828464
-When I'm configuring in a Django app, what is the purpose of the djcelery models? Right --now tables get created, but nothing flows into them. Is this the database backend
---- replacement for Redis and RabbitMQ? Or is it something else?
- Several database tables are used:
* Monitoring
When you use the django-admin monitor the cluster state is written
to the TaskState and WorkerState tables.
@ask
ask / block-eventlet.py
Created October 3, 2011 15:41
https://gist.github.com/1258543 re-implemented with http://eventlet.net/
import eventlet
from eventlet import wsgi
from eventlet import sleep
def switch(it):
for item in it:
yield item
sleep(0.0)
@ask
ask / traverse.py
Created August 31, 2011 12:11
traverse celery results
from collections import deque
from celery.result import BaseAsyncResult, TaskSetResult
from celery.task import chord, task, TaskSet
def force_list(l):
if not isinstance(l, (list, tuple)):
return [l]
return l