agronholm

from pathlib import Path

path = Path("MARKER-ö.txt")
path.touch()

path2 = next(p for p in Path.cwd().iterdir() if p.name.startswith("MARKER-"))
path2.chmod(0o777)

agronholm

from __future__ import annotations

__all__ = ["connect", "Connection", "Cursor"]

import sqlite3
from collections.abc import Callable, Sequence
from functools import partial, wraps
from typing import Any

from anyio import CapacityLimiter, to_thread

agronholm

from asyncio import run, sleep

from asyncpg import InterfaceError
from sqlalchemy.ext.asyncio import create_async_engine


async def main():
    engine = create_async_engine("postgresql+asyncpg://postgres:secret@localhost/testdb")
    while True:
        while True:

agronholm

class _BlockingAsyncContextManager(AbstractContextManager):
    _enter_future: Future
    _exit_future: Future
    _exit_event: Event
    _exit_exc_info: Tuple[Optional[Type[BaseException]], Optional[BaseException],
                          Optional[TracebackType]]

    def __init__(self, async_cm: AsyncContextManager[T_co], portal: 'BlockingPortal'):
        self._async_cm = async_cm
        self._portal = portal

agronholm

from concurrent.futures import as_completed

from anyio import start_blocking_portal, sleep


async def long_running_task(index):
    await sleep(1)
    print(f'Task {index} running...')
    await sleep(index)
    return f'Task {index} return value'

agronholm

test.py:13: note: Revealed type is 'Any'
test.py:13: note: 'reveal_type' always outputs 'Any' in unchecked functions
test.py:15: note: Revealed type is 'Any'
test.py:15: note: 'reveal_type' always outputs 'Any' in unchecked functions
test.py:17: note: Revealed type is 'Any'
test.py:17: note: 'reveal_type' always outputs 'Any' in unchecked functions
test.py:20: error: No overload variant of "foo" matches argument type "float"
test.py:20: note: Possible overload variants:
test.py:20: note:     def foo(bar: Union[bytes, str]) -> str
test.py:20: note:     def foo(bar: int) -> int

agronholm

class TypedAttribute(Generic[T_Attr]):
    """
    Generic class used to define typed attributes, for use in :class:`~TypedAttributeContainer`.
    """


class TypedAttributeContainer(metaclass=ABCMeta):
    """
    Base class for classes that wish to provide typed extra attributes.

agronholm

import asyncio
import socket


class DatagramProtocol(asyncio.DatagramProtocol):
    def error_received(self, exc: Exception) -> None:
        print('received error:', exc.__class__, ':', exc)


async def main():

agronholm

import curio


async def main():
    async def cancel():
        await curio.sleep(0.1)
        await task.cancel(blocking=False)

    event = curio.Event()
    task = await curio.current_task()

agronholm

import logging
import socket
from dataclasses import dataclass, field
from email.headerregistry import Address
from email.message import EmailMessage
from email.utils import getaddresses, parseaddr
from ssl import SSLContext
from typing import Optional, Iterable, Callable, Union, List, Dict, Any

from anyio import connect_tcp, fail_after, BlockingPortal, start_blocking_portal