squeaky-pl

registry = {
    'db': 'my database',
    'redis': 'my redis',
    'settings': 'my settings'
}


def inject(func):
    def wrapper(*args, **kwargs):
        injected = {k: registry[dep] for k, dep in func.__annotations__.items()}

squeaky-pl

In [24]: from sqlalchemy.sql.expression import delete

In [26]: delete(User).where(User.id == 4)
Out[26]: <sqlalchemy.sql.dml.Delete object at 0x7f4b8f7f1b00>

In [28]: s = delete(User).where(User.id == 4)

In [32]: c = s.compile(dialect=dialect)

In [33]: str(c)

squeaky-pl

In [1]: from sqlalchemy.ext.declarative import declarative_base

In [2]: Base = declarative_base()

In [3]: from sqlalchemy import Column, Integer, String

In [4]: class User(Base):
   ...:     __tablename__ = 'users'
   ...:     id = Column(Integer, primary_key=True)
   ...:     name = Column(String)

squeaky-pl

   File "/src/pypy/rpython/tool/algo/graphlib.py", line 100, in visit
    visit(edge.target)
   File "/src/pypy/rpython/tool/algo/graphlib.py", line 100, in visit
    visit(edge.target)
   File "/src/pypy/rpython/tool/algo/graphlib.py", line 100, in visit
    visit(edge.target)
   File "/src/pypy/rpython/tool/algo/graphlib.py", line 100, in visit
    visit(edge.target)
   File "/src/pypy/rpython/tool/algo/graphlib.py", line 100, in visit
    visit(edge.target)

squeaky-pl

import asyncio as aio
import signal
import os
from functools import partial

from datetime import datetime, timedelta

initial_alerts = {
 1: datetime(2015, 12, 12, 12, 13, 4),
 2: datetime(2015, 8, 16, 12, 14, 5)

squeaky-pl

import asyncio as aio
import signal
import os
from functools import partial

from datetime import datetime, timedelta

initial_alerts = {
 1: datetime(2015, 12, 12, 12, 13, 4),
 2: datetime(2015, 8, 16, 12, 14, 5)

squeaky-pl

import asyncio as aio
import signal
import os

from datetime import datetime, timedelta

alerts = {
 1: datetime(2015, 12, 12, 11, 57, 4),
 2: datetime(2015, 8, 16, 11, 58, 5)
}

squeaky-pl

"""SQLAlchemy session integration for asyncio.

This is hazardous and fragile material. Please don't make changes here unless
you really know what you are doing. You need to understand how threads work
and why SQLAlchemy does not work with asyncio out of the box.

For the reference consult following articles:
http://docs.sqlalchemy.org/en/latest/orm/contextual.html#thread-local-scope
https://docs.python.org/3/library/asyncio-eventloop.html#executor
https://docs.python.org/3/library/concurrent.futures.html#threadpoolexecutor

squeaky-pl

PyPy3 nightly server & client
-----------------------------

1KiB

pypy3 client.py --num 10000 --workers 10 --msize 1000
will connect to: ('127.0.0.1', 25000)
Sending 10000 messages
Sending 10000 messages
Sending 10000 messages

squeaky-pl

# Copied with minimal modifications from curio
# https://github.com/dabeaz/curio

from concurrent.futures import ProcessPoolExecutor

import argparse
from socket import *
import time
import sys