相关疑难解决方法(0)

试图用语句和上下文管理器来理解python

我是新手,我只是想了解这个with陈述.我明白它应该替换try/ exceptblock.

现在假设我做了这样的事情:

try:
   name='rubicon'/2 # to raise an exception
except Exception as e:
   print "no not possible"
finally:
   print "Ok I caught you"
Run Code Online (Sandbox Code Playgroud)

如何用上下文管理器替换它?

python contextmanager

16
推荐指数
5
解决办法
9387
查看次数

有没有办法使用 atexit 运行 asyncio 协程?

我正在使用`discord.py 开发一个机器人。机器人创建/删除多个频道,并连接到 SQLite 数据库。如果机器人崩溃,我希望它崩溃

  1. 销毁它创建的所有临时语音通道。
  2. 断开与 SQL 数据库的连接。

这是关闭协程:

async def shutdown(self):
    print("Shutting down Canvas...")
    for ch in self.active_channels:
        await client.delete_channel(ch)
    self.db.close()
Run Code Online (Sandbox Code Playgroud)

我尝试过的事情:

# Canv is the interface between the bot and the data we're collecting
atexit.register(canv.shutdown) 
bot.run(TOKEN)
Run Code Online (Sandbox Code Playgroud)
try:
    bot.loop.run_until_complete(bot.start(TOKEN))
except KeyboardInterrupt or InterruptedError:
    bot.loop.run_until_complete(canv.shutdown())
finally:
    bot.loop.close()
Run Code Online (Sandbox Code Playgroud)
from async_generator import asynccontextmanager

@asynccontextmanager
async def cleanup_context_manager():
    try:
        yield
    finally:
        await canv.shutdown()

with cleanup_context_manager():
    bot.run(TOKEN)
Run Code Online (Sandbox Code Playgroud)

这些都没有运行canv.shutdown(),这是一个asyncio.coroutine. 如何确保此代码在每种类型的出口上运行?

我用这篇文章来获取一些信息,我认为它最接近我想要的。

python asynchronous python-asyncio discord.py

6
推荐指数
2
解决办法
1713
查看次数

如何在python-trio中的KeyboardInterrupt之后清理连接

我的班级在连接到服务器时应立即发送登录字符串,然后在会话结束后应发送退出字符串并清理套接字。下面是我的代码。

import trio

class test:

    _buffer = 8192
    _max_retry = 4

    def __init__(self, host='127.0.0.1', port=12345, usr='user', pwd='secret'):
        self.host = str(host)
        self.port = int(port)
        self.usr = str(usr)
        self.pwd = str(pwd)
        self._nl = b'\r\n'
        self._attempt = 0
        self._queue = trio.Queue(30)
        self._connected = trio.Event()
        self._end_session = trio.Event()

    @property
    def connected(self):
        return self._connected.is_set()

    async def _sender(self, client_stream, nursery):
        print('## sender: started!')
        q = self._queue
        while True:
            cmd = await q.get()
            print('## sending to the server:\n{!r}\n'.format(cmd))
            if self._end_session.is_set():
                nursery.cancel_scope.shield = True …
Run Code Online (Sandbox Code Playgroud)

python python-3.x python-3.5 python-3.6 python-trio

3
推荐指数
1
解决办法
408
查看次数