njsmith

class EnterTimeout:
    def __init__(self, async_cm, timeout):
        self._async_cm = async_cm
        self._timeout = timeout
        
    async def __aenter__(self):
        with fail_after(self.timeout):
            return await self._async_cm.__aenter__()
        
    async def __aexit__(self, *args):

njsmith

import trio
import sys
import time
import httpx
from outcome import Error
import traceback

# Can't use PySide2 currently because of
# https://bugreports.qt.io/projects/PYSIDE/issues/PYSIDE-1313
from PyQt5 import QtCore, QtWidgets

guilledk

import math
import trio

class AsyncQueue:

    def __init__(self):
        self.history = []

        self.inport, self.outport = trio.open_memory_channel(math.inf)

maxidorius

Notes on privacy and data collection of Matrix.org

---

This version of the document is no longer canonical. You can find the canonical version hosted at Gitlab and Github.

PART 2 IS OUT, INCLUDING THE DISCLOSURE OF A GLOBAL FEDERATION DATA LEAK, AND THE ANATOMY OF A GDPR DATA REQUEST HANDLED BY MATRIX.ORG. SEE THE REPOS ABOVE.

tshirtman

from kivy.factory import Factory as F
from kivy.app import App
from kivy.lang import Builder


KV = '''
<ButtonLayout>:
    on_press: print("here or there")

    Label:

njsmith

class UniversalQueue:
    def __init__(self, *args, **kwargs):
        self._queue = trio.Queue(*args, **kwargs)
        self._portal = trio.BlockingTrioPortal()
        
    async def trio_get(self):
        return await self._queue.trio_get()
        
    def thread_get(self):
        return self._portal.run(self._queue.trio_get)

Nadrieril

# I used this shell.nix to build LineageOS 13.0 for my maguro (Samsung Galaxy Nexus GSM) phone
# The build instructions for normal Linuxes are here: https://wiki.lineageos.org/devices/maguro/build
# For NixOS, follow those instructions but skip anything related to installing packages
# Detailed instructions:
#   cd into an empty directory of your choice
#   copy this file there
#   in nix-shell:
#    $ repo init -u https://github.com/LineageOS/android.git -b cm-13.0
#    $ repo sync
#    $ source build/envsetup.sh

nonsleepr

import trio

from multiplexer import Multiplexer

async def reader(mx, key, timeout=100):
    print(f'Waiting for "{key}"...')
    try:
        with trio.fail_after(timeout):
            value = await mx[key]
            print(f'Got value "{value}" for key {key}')

vodik

import asyncio
import collections
from collections.abc import AsyncGenerator


class Subject(AsyncGenerator):
    def __init__(self, *, loop=None):
        if loop is None:
            self._loop = asyncio.get_event_loop()
        else:

timkpaine

'''

To run:
python -m bokeh serve iex.py

'''



import io