import time
from fastapi import FastAPI, UploadFile, Request

app = FastAPI()


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile) -> dict:
    print(len(await file.read()))
    return {"filename": file.filename}


@app.post("/streamfile/")
async def create_file(request: Request):
    chunks: list[bytes] = []
    async for chunk in request.stream():
        chunks.append(chunk)
    print(len(chunks))
    return request.headers.get("filename")


import uvicorn
uvicorn.run(app)

from starlette.formparsers import MultiPartParser

MultiPartParser.max_file_size = 20 * 1024 * 1024

import tempfile

# 修改临时文件目录
tempfile.tempdir = "path/to/tempdir/here"

from starlette.formparsers import MultiPartParser

# 修改该配置，使所有文件都会直接存放到硬盘中
MultiPartParser.max_file_size = 0 

from fastapi import FastAPI, Request
from multipart import PushMultipartParser, MultipartSegment, parse_options_header

app = FastAPI()


@app.post("/streamfile/")
async def create_file(request: Request):
    content_type = request.headers.get("content-type")
    real_content_type, boundary_dict = parse_options_header(content_type)
    assert real_content_type == "multipart/form-data"

    filename: str = ""
    file_length: int = 0
    stream_gen = request.stream()

    with PushMultipartParser(boundary_dict["boundary"]) as parser:
        while not parser.closed:
            chunk = await stream_gen.__anext__()
            for result in parser.parse(chunk):
                if isinstance(result, MultipartSegment):
                    print(f"== Start of segment: {result.name}")
                    if result.filename:
                        filename = result.filename
                        print(f"== Client-side filename: {result.filename}")
                    for header, value in result.headerlist:
                        print(f"{header}: {value}")
                elif result:  # Result is a non-empty bytearray
                    file_length += len(result)
                else:  # Result is None
                    print(f"== End of segment")

    return {"filename": filename, "length": file_length}


import uvicorn
uvicorn.run(app)

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")

import uvicorn
uvicorn.run(app)

import asyncio
from aiohttp import web

routes = web.RouteTableDef()
routes.static("/static", "./static")

app = web.Application()
app.add_routes(routes)

def main(port=8000):
    loop = asyncio.new_event_loop()
    web.run_app(app, port=port, loop=loop)

if __name__ == '__main__':
    main()

# 0.39版本后新增了if-range的处理，代码会更多，这里以0.38版本代码为准
class FileResponse(Response):
    ...
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        ...
        await send(
            {
                "type": "http.response.start",
                "status": self.status_code,
                "headers": self.raw_headers,
            }
        )
        if scope["method"].upper() == "HEAD":
            await send({"type": "http.response.body", "body": b"", "more_body": False})
        else:
            # >>>主要逻辑
            async with await anyio.open_file(self.path, mode="rb") as file:
                more_body = True
                while more_body:
                    chunk = await file.read(self.chunk_size)
                    more_body = len(chunk) == self.chunk_size
                    await send(
                        {
                            "type": "http.response.body",
                            "body": chunk,
                            "more_body": more_body,
                        }
                    )
        if self.background is not None:
            await self.background()

class FileResponse(StreamResponse):
    ...
    async def _sendfile_fallback(
        self, writer: AbstractStreamWriter, fobj: IO[Any], offset: int, count: int
    ) -> AbstractStreamWriter:

        chunk_size = self._chunk_size
        loop = asyncio.get_event_loop()

        await loop.run_in_executor(None, fobj.seek, offset)

        chunk = await loop.run_in_executor(None, fobj.read, chunk_size)
        while chunk:
            await writer.write(chunk)
            count = count - chunk_size
            if count <= 0:
                break
            chunk = await loop.run_in_executor(None, fobj.read, min(chunk_size, count))

        await writer.drain()
        return writer

    async def _sendfile(
        self, request: "BaseRequest", fobj: IO[Any], offset: int, count: int
    ) -> AbstractStreamWriter:
        writer = await super().prepare(request)
        assert writer is not None

        if NOSENDFILE or self.compression:
            return await self._sendfile_fallback(writer, fobj, offset, count)

        loop = request._loop
        transport = request.transport
        assert transport is not None

        try:
            await loop.sendfile(transport, fobj, offset, count)
        except NotImplementedError:
            return await self._sendfile_fallback(writer, fobj, offset, count)

        await super().write_eof()
        return writer

from fastapi import Response, FastAPI


async def file_route(file_name: str) -> Response:
    resp = Response()
    resp.headers["X-Sendfile"] = f"/private_files/{file_name}"
    return resp


app = FastAPI()
app.add_route("/api/file/{file_name}", file_route, methods=["GET"])


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app)

import asyncio


def get_event_loop() -> asyncio.AbstractEventLoop:
    """get event loop in runtime"""
    if sys.version_info >= (3, 7):
        return asyncio.get_running_loop()

    loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
    if not loop.is_running():
        raise RuntimeError("no running event loop")
    return loop

import asyncio


class Demo(object):
    def __init__(self, loop=None):
        if loop is None:
            loop = asyncio.get_event_loop()
        else:
            loop = loop

import asyncio
import uvicorn
from fastapi import FastAPI


class BadClient(object):
    def __init__(self) -> None:
        self._loop = asyncio.get_event_loop()

    async def fake_request(self) -> None:
        await self._loop.create_task(asyncio.sleep(1))
        return None


app = FastAPI()
client = BadClient()


@app.get("/")
async def demo() -> str:
    await client.fake_request()
    return "Hello World!"


uvicorn.run(app)

"""
省略一些代码...
"""
app = FastAPI()

@app.on_event("startup")
async def startup() -> None:
    app.client = BadClient()


@app.get("/")
async def demo(request: Request) -> str:
    await request.app.client.fake_request()
    return "Hello World!"


uvicorn.run(app)

import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def demo() -> str:
    return "Hello World!"


uvicorn.run(app)

import requests
import uvicorn
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def demo() -> str:
    status_code: int = requests.get("https://baidu.com").status_code
    return f"Hello World!, status code:{status_code}"


uvicorn.run(app)

# 位于:Lib/asyncio/base_events.py
    async def getaddrinfo(self, host, port, *, family=0, type=0, proto=0, flags=0):
        if self._debug:
            getaddr_func = self._getaddrinfo_debug
        else:
            getaddr_func = socket.getaddrinfo

        return await self.run_in_executor(
            None, getaddr_func, host, port, family, type, proto, flags)

import asyncio
from concurrent.futures import ThreadPoolExecutor

loop = asyncio.get_event_loop()
loop.set_default_executor(ThreadPoolExecutor(10))

import asyncio
import httpx


async def main() -> None:
    async with httpx.AsyncClient() as client:
        response = await client.get("http://example.com")
        print(response.json())


asyncio.run(main())

import asyncio
import time


async def sub_task1() -> None:
    time.sleep(1)


async def sub_task2() -> None:
    print("Hello!")
    await asyncio.sleep(1)
    print("Hi")
    await asyncio.sleep(1)


async def sub_task3() -> None:
    await sub_task1()

async def sub_task4() -> None:
    t = asyncio.create_task(sub_task1())
    await t

async def main() -> None:
    t1 = asyncio.create_task(sub_task1())
    t2 = asyncio.create_task(sub_task2())
    t3 = asyncio.create_task(sub_task3())
    t4 = asyncio.create_task(sub_task4())
    await asyncio.gather(t1, t2, t3, t4)


asyncio.run(main(), debug=True)

import asyncio


async def main() -> None:
    pass


loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main())
finally:
    loop.close()

import asyncio


async def main() -> None:
    pass


asyncio.run(main())

import asyncio
from threading import Thread

event_loop = asyncio.get_event_loop()

async def do_something():
    print("Hi")

def new_event_loop():
    event_loop.run_forever() # 线程内阻塞

t = Thread(target=new_event_loop).start()

for i in range(100):
    asyncio.run_coroutine_threadsafe(do_something(), loop=event_loop)

from fastapi import FastAPI, Depends

app = FastAPI()


###########
#  第一块  #
###########
class FakerDB(object):
    def __init__(self) -> None:
        self._initialized = False
    async def create(self) -> None:
        self._initialized = True

    async def get_user_info(self) -> dict:
        if not self._initialized:
            raise RuntimeError("FakerDB not initialized")
        return {
            "id": 1,
            "name": "so1n",
            "age": 18,
        }


###########
#  第二块  #
###########
_db = None

def get_user_db() -> FakerDB:
    global _db
    if _db is None:
        _db = FakerDB()
    return _db


###########
#  第三块  #
###########
@app.on_event("startup")
async def startup_event(user_db: FakerDB = Depends(get_user_db)):
    await user_db.create()


@app.get("/api/user/info")
async def login(user_db: FakerDB = Depends(get_user_db)):
    return await user_db.get_user_info()


import uvicorn

uvicorn.run(app)

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

import time
import uvicorn
from fastapi import FastAPI, Depends, Query
from asyncio import Queue, sleep

app = FastAPI()


###########
#  第一块  #
###########
class FakerDB(object):
    async def get_user_info(self) -> dict:
        return {
            "id": 1,
            "name": "so1n",
            "age": 18,
        }


###########
#  第二块  #
###########
_db_pool = Queue()


async def get_user_db() -> FakerDB:
    db = await _db_pool.get()
    try:
        yield db
    finally:
        _db_pool.put_nowait(db)



###########
#  第三块  #
###########
@app.on_event("startup")
async def start_up() -> None:
    # 由于uvicorn的特性，需要在这里重新创建Queue，不然它们将不会被同一个event loop调用
    global _db_pool
    _db_pool = Queue()

    for _ in range(3):
        _db_pool.put_nowait(FakerDB())


@app.get("/api/user/info")
async def login(uid: int = Query(...), user_db: FakerDB = Depends(get_user_db)):
    print(f"Hello! {uid}, access time {time.time()}")
    await sleep(3)  # 模拟其它io
    return await user_db.get_user_info()


uvicorn.run(app)

import httpx
import asyncio


async def main():
    async with httpx.AsyncClient() as client:
        tasks = [client.get(f"http://127.0.0.1:8000/api/user/info?uid={i}") for i in range(5)]
        await asyncio.gather(*tasks)


asyncio.run(main())

import json
import time
from redis.asyncio import Redis
from httpx import AsyncClient
from typing import Optional

_namespace = "example:info"

class InfoService(object):
    def __init__(self, *, redis: Redis, client: AsyncClient) -> None:
        self.redis = redis
        self.client = client

    async def get(self, url: str, cache_timeout: int = 9) -> dict:
        redis_result: Optional[str] = await self.redis.get(f"{_namespace}:{url}")
        if not redis_result:
            async with self.client as client:
                response = await client.get(url)
                status_code = response.status_code
                timestamp = int(time.time())
            result = {"timestamp": timestamp, "status_code": status_code}
            await self.redis.set(f"{_namespace}:{url}", json.dumps(result), ex=cache_timeout)
        else:
            result = json.loads(redis_result)
        return result

from dependency_injector import containers, providers
from redis.asyncio import Redis
from httpx import AsyncClient
from services import InfoService


class Container(containers.DeclarativeContainer):
    config = providers.Configuration(yaml_files=["config.yaml"])

    redis = providers.Singleton(
        Redis,
        host=config.redis.host,
        port=config.redis.port,
    )
    client = providers.Singleton(
        AsyncClient,
        timeout=config.http.timeout,
    )
    info_service = providers.Factory(
        InfoService,
        redis=redis,
        client=client,
    )

from fastapi import APIRouter, Depends, Body
from dependency_injector.wiring import inject, Provide
from services import InfoService
from containers import Container


router = APIRouter()

@router.post("/")
@inject
async def demo(
    url: str = Body(..., embed=True),
    cache_timeout: int = Depends(Provide[Container.config.default.cache_timeout]),
    info_service: InfoService = Depends(Provide[Container.info_service]),
):
    return await info_service.get(url, cache_timeout)

from fastapi import FastAPI
from dependency_injector.wiring import inject, Provide

from containers import Container
import endpoints


def create_app() -> FastAPI:
    container = Container()

    app = FastAPI()
    app.container = container
    app.include_router(endpoints.router)

    @app.on_event("startup")
    async def start_up_1() -> None:
        result = await app.container.redis().info()
        if result:
            print(f"Redis: {id(app.container.redis())} connect success 1")

    @app.on_event("startup")
    @inject
    async def start_up_2(_container: Container = Provide[Container]) -> None:
        result = await _container.redis().info()
        if result:
            print(f"Redis: {id(_container.redis())} connect success 2")

    container.wire(modules=[__name__, endpoints])
    return app


app = create_app()

from dependency_injector.wiring import Provide
from typing import TypeVar, Type
_T = TypeVar("_T")

def MyProvide(a: Type[_T]) -> _T:
    return Provide[a]

@app.on_event("startup")
@inject
async def start_up_2(_container = MyProvide(Container)) -> None:
    result = await _container.redis().info()
    if result:
        print(f"Redis: {id(_container.redis())} connect success 2")

class Lock(object):
    def __init__(self, name: str, client: aetcd.Client, ttl: int) -> None:
        self._prefix = name.encode()
        self._name = self._prefix + b'/' + uuid1().hex.encode()
        self._client = client
        self._ttl = ttl
        self.lease = None
        self._watch_dog: Optional[asyncio.Task]= None

    async def acquire(self) -> bool:
        # 1.创建续约
        self.lease = await self._client.lease(self._ttl)
        # 2.写入数据
        create_result = await self._client.put(self._name, b"", lease=self.lease)
        self_revision = create_result.header.revision
        # 3.查询数据
        range_result = await self._client.get_prefix(self._prefix, sort_target="create")
        create_revision = range_result.kvs[0].create_revision
        if create_revision == self_revision:
            return True

        # 4.查询需要监听的key
        watch_key = b""
        for index, item in enumerate(range_result.kvs):
            if item.create_revision == self_revision:
                watch_key = range_result.kvs[index - 1].key

        # 5.监听Key
        watch = await self._client.watch(
            watch_key,
            kind=aetcd.rtypes.EventKind.DELETE,
        )
        async for event in watch:
            return True

class Lock(object):
    ...
    async def __aenter__(self) -> "Lock":
        self._watch_dog = asyncio.create_task(self.watch_dog())
        await self.acquire()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        if self._watch_dog and not self._watch_dog.done():
            self._watch_dog.cancel()
        await self.release()

    async def watch_dog(self):
        while True:
            if self.lease is not None:
                try:
                    await self.refresh()
                except ValueError:
                    pass
            await asyncio.sleep(self._ttl / 3)
    async def refresh(self):
        """Refresh the time to live on this lock."""
        if self.lease is not None:
            return await self.lease.refresh()

        raise ValueError(f'no lease associated with this lock: {self._name!r}')

    async def release(self) -> None:
        await self._client.delete(self._name)

def my_print(msg: str):
    print(f"Timestamp:{time.time()} Task:{id(asyncio.current_task())}, {msg}")


async def sub(client: aetcd.Client, cnt: int) -> None:
    my_print(f"cnt:{cnt} wait")
    async with Lock("demo", client, 1):
        my_print(f"cnt:{cnt} run")
        await asyncio.sleep(2)
    my_print(f"cnt:{cnt} done")


async def main() -> None:
    client = aetcd.Client()
    await client.delete_prefix(b"demo")
    tasks = []
    tasks.append(asyncio.create_task(sub(client, 1)))
    await asyncio.sleep(0.1)
    tasks.append(asyncio.create_task(sub(client, 2)))
    await asyncio.sleep(0.1)
    tasks.append(asyncio.create_task(sub(client, 3)))
    await asyncio.gather(*tasks)


if __name__ == "__main__":
    asyncio.run(main())

class Manager(object):
    def __init__(self, client: Redis, proxy_num: int = 8, timeout: int = 1):
        self._proxy_num = proxy_num
        self._proxies: List[Proxy] = [Proxy(str(i), self) for i in range(proxy_num)]

        self.client = client
        self.timeout = timeout
        self.listener_dict: Dict[str, asyncio.Queue[asyncio.Future]] = {}

    def listen(self, lock_name: str) -> asyncio.Future:
        """创建一个future加入到队列中，并把future返回给调用者"""
        if lock_name not in self.listener_dict:
            self.listener_dict[lock_name] = asyncio.Queue()

        f = asyncio.Future()
        self.listener_dict[lock_name].put_nowait(f)
        return f

    def empty(self, lock_name: str) -> bool:
        """判断队列是否为空"""
        if lock_name not in self.listener_dict:
            return True
        return self.listener_dict[lock_name].empty()

    def start(self) -> None:
        for proxy in self._proxies:
            proxy.start()

    def stop(self) -> None:
        for proxy in self._proxies:
            proxy.stop()

    def get_random_proxy_name(self) -> str:
        return self._proxies[random.randint(0, len(self._proxies) - 1)].name

class Proxy(object):
    def __init__(self, name: str, manager: "Manager") -> None:
        self.name = name
        self._manager = manager
        self._task: Optional[asyncio.Task] = None

    async def run(self) -> None:
        while True:
            # 通过blpop命令等待Redis的推送
            result: Optional[List[bytes]] = await self._manager.client.blpop([self.name], self._manager.timeout)
            if result is None:
                continue
            # 判断监听器释放存在
            listen_name = result[1].decode()
            if listen_name not in self._manager.listener_dict:
                logging.error(f"{listen_name} not in listener_dict")
                continue

            while True:
                # 如果队列为空，忽略
                if self._manager.listener_dict[listen_name].empty():
                    break
                f = await self._manager.listener_dict[listen_name].get()
                if f.done():
                    # 如果future已经执行完毕，那么当前任务可能获取到锁了
                    continue
                f.set_result(True)
                break

    def start(self) -> None:
        if self._task:
            raise RuntimeError(f"{self.name} already started")
        self._task = asyncio.create_task(self.run())

    def stop(self) -> None:
        if not self._task or self._task.done():
            raise RuntimeError(f"{self.name} not started")
        self._task.cancel()

class Lock(object):
    ##################
    # 暴露给外部的方法 #
    ##################
    def __init__(self):
        pass

    async def __aenter__(self):
        pass
    
    async def __aexit__(self):
        pass
    
    ################
    # token处理逻辑 #
    ###############
    @staticmethod
    def _new_token() -> str:
        pass

    @property
    def _token(self) -> Optional[str]:
        pass

    @_token.setter
    def _token(self, token: Optional[str]) -> None:
        pass

    ##############
    # 锁的处理逻辑 #
    ##############
    async def _acquire(self):
        pass

    async def _extend(self):
        pass

    async def _release(self):
        pass

class BaseLock(object):
    def __init__(
            self,
            client: Redis,
            manager: Manager,
            name: str,
            timeout: int = 9,
    ) -> None:
        self._name = name
        self._manager = manager
        self._client = client
        self._timeout = timeout
        self.register_scripts()

    async def __aenter__(self):
        if await self.acquire():
            self._cancel_watch_dog()
            self._watch_dog = asyncio.create_task(self._watch())
            return self
        raise LockError("Unable to acquire lock within the time specified")

    async def __aexit__(self, exc_type, exc_value, traceback):
        self._cancel_watch_dog()
        await self._release()

class BaseLock(object):
    ...
    lua_acquire = None
    LUA_ACQUIRE_SCRIPT = ""  # 省略lua 脚本 

    async def acquire(self) -> bool:
        if self._token:
            raise LockError("Lock already acquired")
        token = self._new_token()
        proxy_name = self._manager.get_random_proxy_name()
        listener: Optional[asyncio.Future] = None
        ttl: Optional[int] = self._timeout

        while True:
            if listener or self._manager.empty(self._name):
                # 为了保证公平锁的公平，只有在listener或者队列为空才可以进入尝试获取锁的逻辑。
                ttl = await self._do_acquire(token, proxy_name)
                if not ttl:
                    self._token = token
                    if listener and not listener.done():
                        # 可能是ttl超时而重新获取锁并获取成功了，所以需要关闭监听器
                        listener.set_result(True)
                    return True
            # 获取不到锁，先获取监听器，然后再通过监听器和超时决定什么时候再获取锁。
            if not listener or listener.done():
                listener = self._manager.listen(self._name)
            # 超时的值不仅是ttl，它应该考虑网络波动，所以应该比ttl再大一些
            ttl = ttl / 3 if (ttl / 3) < 1 else 1
            await asyncio.wait([listener], timeout=ttl)

    async def _do_acquire(self, token: str, proxy_name: str) -> Optional[int]:
        timeout = int(self._timeout * 1000)
        return await self.lua_acquire(
            keys=[self._name], args=[token, proxy_name, timeout], client=self._client
        )

class BaseLock(object):
    ...
    lua_release = None
    LUA_RELEASE_SCRIPT = ""  # 省略lua脚本
    async def _release(self) -> None:
        token = self._token
        if token is None:
            raise LockError("Cannot release an unlocked lock")

        result: Optional[int] = await self._do_release(token)
        if result != 0:
            self._token = None
        if result is None:
            raise LockNotOwnedError("Cannot release a lock that's no longer owned")

    async def _do_release(self, token: str) -> Optional[int]:
        return await self.lua_release(
            keys=[self._name], args=[token], client=self._client
        )

async def print_info(cnt: int) -> None:
    print(f"Task:{id(asyncio.current_task())}, run cnt:{cnt}")


async def demo(manager: Manager, client: Redis) -> None:
    await print_info(0)
    async with BaseLock(client, manager, "demo", timeout=1):
        await print_info(1)
        await asyncio.sleep(1)


async def main():
    _redis = Redis()
    manager = Manager(client=_redis)
    manager.start()
    await asyncio.gather(*[demo(manager, _redis) for _ in range(3)])
    manager.stop()


if __name__ == '__main__':
    asyncio.run(main())

import asyncio
from redis.asyncio import Redis


async def main():
    _redis = Redis()
    _lock = _redis.lock("demo", timeout=3)
    async with _lock:
        print("get lock 1")
        async with _lock:
            print("get lock 2")


asyncio.run(main())

import asyncio
from contextvars import ContextVar

ctx = ContextVar("aaa")

async def sub(cnt):
    parent_value = ctx.get()
    ctx.set(cnt)
    print(f"Task:{id(asyncio.current_task())}, now_ctx_value:{cnt}, parent_ctx_value:{parent_value}")


async def main():
    ctx.set("bbb")
    await asyncio.gather(sub("1"), sub("2"))
    print("main ctx:", ctx.get())

if __name__ == "__main__":
    asyncio.run(main())

from contextvars import ContextVar

_lock_ctx: ContextVar[dict] = ContextVar("lock_ctx", default={})


class AllowNestedLock(object):
    ...
    @staticmethod
    def _new_token() -> str:
        return str(uuid1().hex) + ":" + str(id(asyncio.current_task()))

    @property
    def _token(self) -> Optional[str]:
        token: Optional[str] = _lock_ctx.get().get("token", None)
        if token:
            token += ":" + str(id(asyncio.current_task()))
        return token

    @_token.setter
    def _token(self, token: Optional[str]) -> None:
        ctx_dict = _lock_ctx.get()
        if token:
            token = token.split(":")[0]
        ctx_dict["token"] = token

class AllowNestedLock(object):
    ...
    async def __aenter__(self):
        if await self.acquire():
            ctx_dict = _lock_ctx.get()
            if "watch_dog" not in ctx_dict:
                ctx_dict["watch_dog"] = id(self)
                self._cancel_watch_dog()
                self._watch_dog = asyncio.create_task(self._watch(self._token))
            return self
        raise LockError("Unable to acquire lock within the time specified")

    async def __aexit__(self, exc_type, exc_value, traceback):
        if _lock_ctx.get().get("watch_dog", 0) == id(self):
            self._cancel_watch_dog()
            _lock_ctx.get().pop("watch_dog", None)
        await self._release()

async def print_info(cnt: int) -> None:
    print(f"Task:{id(asyncio.current_task())}, run cnt:{cnt}")


async def demo(manager: Manager, client: Redis) -> None:
    await print_info(0)
    async with AllowNestedLock(client, manager, "demo", timeout=1):
        await print_info(1)
        async with AllowNestedLock(client, manager, "demo", timeout=1):
            await print_info(2)
            await asyncio.sleep(1)


async def main():
    _redis = Redis()
    manager = Manager(client=_redis)
    manager.start()
    await asyncio.gather(*[demo(manager, _redis) for _ in range(3)])
    manager.stop()


if __name__ == '__main__':
    asyncio.run(main())

async def demo(manager: Manager, client: Redis) -> None:
    await print_info(0)
    lock = AllowNestedLock(client, manager, "demo", timeout=1)
    async with lock:
        await print_info(1)
        async with lock:
            await print_info(2)
            await asyncio.sleep(1)

class Manager(object):
    def listen(self, lock_name: str, lock_mode: str) -> asyncio.Future:
        if lock_name not in self.listener_dict:
            self.listener_dict[lock_name] = deque()

        f = asyncio.Future()
        self.listener_dict[lock_name].append((lock_mode, f))
        return f

class Proxy(object):

    async def run(self) -> None:
        while True:
            result: Optional[List[bytes]] = await self._manager.client.blpop([self.name], self._manager.timeout)
            if result is None:
                continue
            listen_name = result[1].decode()
            if listen_name not in self._manager.listener_dict:
                logging.error(f"{listen_name} not in listener_dict")
                continue
            listener = self._manager.listener_dict[listen_name]
            while True:
                if not len(listener):
                    self._manager.listener_dict.pop(listen_name, None)
                    break
                lock_mode, f = listener.popleft()
                if f.done():
                    continue
                f.set_result(True)

                if lock_mode == "read" and listener and listener[0][0] == "read":
                    continue
                else:
                    break