标签: python-asyncio

客观的：

我正在尝试同时抓取多个网址。我不想同时发出太多请求，因此我使用此解决方案来限制它。

问题：

正在为所有任务发出请求，而不是一次针对有限数量的任务。

精简代码：

async def download_all_product_information():
    # TO LIMIT THE NUMBER OF CONCURRENT REQUESTS
    async def gather_with_concurrency(n, *tasks):
        semaphore = asyncio.Semaphore(n)

        async def sem_task(task):
            async with semaphore:
                return await task

        return await asyncio.gather(*(sem_task(task) for task in tasks))

    # FUNCTION TO ACTUALLY DOWNLOAD INFO
    async def get_product_information(url_to_append):
        url = 'https://www.amazon.com.br' + url_to_append

        print('Product Information - Page ' + str(current_page_number) + ' for category ' + str(
            category_index) + '/' + str(len(all_categories)) + ' in …

我试图在 FastApi 路由中运行子进程，但执行结果为NotImplementedError. 我读过有关该问题的类似问题：

为什么我在 Windows 上使用 async 和 wait 时会收到 NotImplementedError 错误？

Asyncio.create_subprocess_exec NotImplementedError - Fastapi 后台任务

但他们似乎没有任何可行的解决方案。

我的 FastApi 路线如下所示：

@app.get("/test/subprocess")
async def subprocess_test():
    parsed_json = await(probe_video_file(Path(r"G:\ffmpeg testing\ffmpeg\ffprobe.exe"),
                                         Path(r"G:\ffmpeg testing\input_file\test_file.wmv")))
    print(parsed_json)
    return parsed_json

当我导航到此路径时，会引发异常并且代码崩溃。该probe_video_file函数内部有一个子进程调用，如下所示：

async def probe_video_file(ffprobe_path: Path, file_to_probe: Path) -> dict:
    """
    Probes a video file with FFprobe.
    :param ffprobe_path: Path to ffprobe executable
    :param file_to_probe: Path to file to probe
    :returns Parsed JSON dict of the output
    """
    args = …

这些功能有什么区别

1. create_task和call_soon
2. call_soon_threadsafe和run_coroutine_threadsafe

create_task 与 call_soon

这两个函数调度协程的执行，不是线程安全的。
create_task返回一个Task对象，和call_soon返回Handle一个对象，但是除此之外还有什么区别，特别是在用法上？

call_soon_threadsafe 与 run_coroutine_threadsafe

这两个函数应该从与运行事件循环的线程不同的操作系统线程调用。在大多数情况下，我可以使用其中任何一个。

我正在并行执行一堆异步调用，如下所示：

txs = await asyncio.gather(*[fetch_tx_details(s["signature"]) for s in sigs])

这些调用有时可能会失败，因此我用退避来装饰每个调用，如下所示：

@retry_with_backoff(10)
async def fetch_tx_details(sig):
    # stuff

其中退避定义如下：

def retry_with_backoff(retries=5, backoff_in_ms=100):
    def wrapper(f):
        @functools.wraps(f)
        async def wrapped(*args, **kwargs):
            x = 0
            while True:
                try:
                    return await f(*args, **kwargs)
                except Exception as e:
                    print('Fetch error:', e)

                    if x == retries:
                        raise
                    else:
                        sleep_ms = (backoff_in_ms * 2 ** x +
                                    random.uniform(0, 1))
                        time.sleep(sleep_ms / 1000)
                        x += 1
                        print(f'Retrying {x + 1}/{retries}')

        return wrapped

    return wrapper

我遇到的问题是，如果任何调用失败，它们将按顺序重试，而不是并行重试。例如，我正在尝试 1000 个调用，100 个失败 - 我现在执行了 …

我正在设法通过 Django 将 FCM 通知发送到我的应用程序。这是我的推送通知功能：

import firebase_admin
from firebase_admin import credentials, messaging

cred = credentials.Certificate("service-key.json")
firebase_admin.initialize_app(cred)

def send_push(title, msg, registration_token, dataObject=None):
    try:
        message = messaging.MulticastMessage(
            notification=messaging.Notification(
                title=title,
                body=msg,
            ),
            data=dataObject,
            tokens=registration_token,
        )

        response = messaging.send_multicast(message)
    except messaging.QuotaExceededError:
        raise f'Exceeding FCM notifications quota'

现在，我将在视图内发送通知：

class AdminChangeRole(viewsets.Viewset):
    serializer_class = SomeSerializer
    permission_classes = [IsAdminOnly]
    def post(self, request, *args, **kwargs):
        # code that will change the role of a user
        send_push("Role changed", f"You role was set to {self.role}", fcm_registration_token)
        # return Response

现在，在序列化器中发布一些数据并保存它之后。我希望向用户发送通知。但是，我想知道我是否正确处理了这个问题，包括并行 2500 …

我遇到过这个问题，虽然它一定是一个常见问题，但我看不到任何解决方案。所以，也许我在这里遗漏了一些东西。

我正在开发具有异步端点和与数据库异步连接的 FastAPI 应用程序。数据库连接作为依赖项传递。我想为所述应用程序编写一些异步测试。

engine = create_async_engine(connection_string, echo=True)

def get_session():
    return sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

@router.post("/register")
async def register(
    user_data: UserRequest,
    authorize: AuthJWT = Depends(),
    async_session: sessionmaker = Depends(get_session),
):
    """Register new user."""
    if authorize.get_jwt_subject():
        raise LogicException("already authorized")

    session: AsyncSession
    async with async_session() as session:
        query = await session.execute(
            select(UserModel).where(UserModel.name == user_data.name)
        )
    ...

我正在使用 AsyncSession 来处理数据库。所以在我的测试中，数据库连接也必须是异步的。

engine = create_async_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
app.dependency_overrides[get_session] = lambda: sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

@pytest.mark.asyncio
async def test_create_user():
    async with engine.begin() as conn: …

我收到此异常：“发生异常：AttributeError __aenter__” in line async with bot:。（我输入了令牌）。请帮忙

import asyncio
import telegram


async def main():
    bot = telegram.Bot("token")
    async with bot:
        print(await bot.get_me())


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

我在这里找到了这段代码：https ://github.com/python-telegram-bot/python-telegram-bot/wiki/Introduction-to-the-API

我最近更新了 SQLAlchemy（带有 [asyncio] 包），1.4.46并在提交时开始出现以下异常：

\n

sqlalchemy.exc.IllegalStateChangeError：此处无法调用方法“commit()”；方法“_connection_for_bind()”已在进行中，这将导致状态意外更改为 <SessionTransactionState.CLOSED: 5>

\n

在更新到新版本之前，它运行良好。

# -*- coding:utf-8 -*-\n\nfrom sqlalchemy import exc, event, text\nfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSession as SQLAlchemyAsyncSession\nfrom sqlalchemy.orm import sessionmaker, Session\nfrom sqlalchemy.ext.asyncio import AsyncEngine\nfrom sqlalchemy.pool import NullPool, Pool\nfrom contextvars import ContextVar\nfrom sanic import Sanic\nimport asyncio\n\n\nclass EngineNotInitialisedError(Exception):\n    pass\n\n\nclass DBSessionContext:\n    def __init__(self, session: Session, commit_on_exit: bool = True) -> None:\n        self.session = session\n        self._query = None\n        self.commit_on_exit = commit_on_exit\n        self.token = None\n\n    async def close(self, exc_type=None, exc_value=None, traceback=None):\n        if …

我正在使用翻译软件，表达上可能存在一些错误，请理解我已经查了网上的教程并按照相应的思路进行操作，但是我遇到了一点问题，我无法使用异步SqlAlchemyc来操作Mysql。

数据库连接

from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from urllib import  parse
import  sys

engine = create_async_engine(Sql_URL)


SessionLocal = sessionmaker(bind=engine,autocommit=False,autoflush=False)
db_session = scoped_session(SessionLocal)

Base = declarative_base()
Base.query = db_session.query_property()

楷模

import asyncio
import datetime

from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey, MetaData
from sqlalchemy.orm import registry
from  sqlalchemy_utils import EmailType,ChoiceType

from app.Fast_blog.database import Base


class user(Base):
    choices = [
        ('0', 'woman'),
        ('1', 'man'),
        ('2', 'NULL')
    ]
    __tablename__ …

考虑以下代码：

主要.py

import asyncio
import websockets


async def echo(websocket):
    async for message in websocket:
        await websocket.send(message)
        print(message)


async def main():
    async with websockets.serve(echo, "localhost", 8765):
        await asyncio.Future()  # run forever

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

其他.py

import asyncio
import json

import websockets

tasks = set()


async def run_job(i):
    await asyncio.sleep(0.)
    print(f"I'm job number {i}")


async def bunch_of_tasks(ws):
    for i in range(10):
        task = asyncio.create_task(run_job(i), name=f'job-{i}')
        tasks.add(task)
        task.add_done_callback(tasks.discard)
        print(f'had a nice sleep! now my value is {i}')
        # await asyncio.sleep(0.)
        await …