agronholm

from functools import wraps


class A:
    @classmethod
    def method(cls):
        print('cls = %s' % cls.__name__)


class B(A):

agronholm

import trio


async def error_after(delay):
    await trio.sleep(delay)
    raise Exception('foo')


async def somegenerator():
    try:

agronholm

import ast
import re
import sys
from importlib.machinery import SourceFileLoader
from importlib.abc import MetaPathFinder
from importlib.util import decode_source, cache_from_source
from typing import Iterable
from unittest.mock import patch

agronholm

from functools import partial
from inspect import iscoroutinefunction

import pytest

import hyperio

try:
    from async_generator import isasyncgenfunction
except ImportError:

agronholm

@pytest.fixture(scope='session', autouse=True)
def sqlalchemy_thread_check(connection):
    """Raise an exception is SQL is executed in the event loop thread in the application code."""
    def check_thread(*args):
        if asyncio._get_running_loop():
            frames = [f for f in extract_stack()[:-1] if 'ebserver' in f.filename]
            if frames:
                raise RuntimeError(f'SQL executed in the event loop thread '
                                   f'(from {frames[-1].filename}:{frames[-1].lineno})')

agronholm

from threading import Thread

counter = 0
threads = []


def increment():
    global counter
    for _ in range(500000):
        counter += 1

agronholm

import gc
import inspect
import threading

import trio
from trio import run_sync_in_worker_thread
from trio.hazmat import (
    spawn_system_task, wait_task_rescheduled, reschedule, current_task, Error, Value, Abort)

agronholm

import inspect
from asyncio import get_event_loop, gather, Task, wait
from concurrent.futures import Executor, ThreadPoolExecutor, ProcessPoolExecutor
from functools import wraps, partial
from typing import Callable, Union, TypeVar, Awaitable, List, Dict

T_Retval = TypeVar('T_Retval')


class MultiError(Exception):

agronholm

import logging

import trio
from wampproto.impl.client.trio import TrioWAMPClient


def procedure_handler(name):
    return 'Hello, {}!'.format(name)

agronholm

[metadata]
name = wampproto
description = WAMP (Web Application Message Protocol) state-machine based protocol implementation
long_description = file: README.rst
author = Alex Grönholm
author_email = [email protected]
url = https://github.com/agronholm/wampproto
license = MIT
license_file = LICENSE
keywords = wamp, websockets