Asyncio.create_subprocess_exec NotImplementedError - Fastapi Bac

ghz 7months ago ⋅ 164 views

Asyncio.create_subprocess_exec NotImplementedError - Fastapi Background Task

I am trying to call asyncio.create_subprocess_exec within a Fastapi background task, but it keeps raising a NotImplementedError. The run_subprocess function works fine when it is ran outside of Fastapi. I am running this in windows using an asyncio loop, not uvloop.

import asyncio
from fastapi import FastAPI, BackgroundTasks

DHCP_SERVER = "1.1.1.1"

app = FastAPI()

@app.get("/")
async def subprocess_test(background_tasks: BackgroundTasks):
  background_tasks.add_task(run_subprocess)

async def run_subprocess():
  proc = await asyncio.create_subprocess_exec(
    'powershell.exe',
    f'Get-Dhcp-Serverv4Scope -ComputerName \"{DHCP_SERVER}\"',
    stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
  )

  stdout, stderr = await proc.communicate()

  if stderr:
    print(stderr)
  else:
    print(stdout)


File ".\subprocess_example.py", line 13 in run_subprocess
  proc = await asyncio.create_subprocess_exec(
File "C:\Python\Python38-32\lib\asyncio\subprocess.py", line 236, in create_subprcess_exec
  transport, protocol = await loop.subprocess_exec(
File "C:\Python\Python38-32\lib\asyncio\base_events.py", line 1615, in subprocess_exec
  transport = await self._make_subprocess_transport(
File "C:\Python\Python38-32\lib\asyncio\base_events.py", line 487, in _make_subprocess_transport
  raise NotImplementedError

Can anyone please help fix this?

Thanks!

Answers

The NotImplementedError you're encountering when trying to call asyncio.create_subprocess_exec on Windows is likely due to the use of the default event loop policy, which might not support subprocesses.

On Windows, the default event loop policy for asyncio does not support subprocesses. However, you can change the event loop policy to asyncio.WindowsProactorEventLoopPolicy, which does support subprocesses. You can set this policy at the beginning of your script to ensure it's used throughout your application.

Here is how you can modify your script to use asyncio.WindowsProactorEventLoopPolicy:

import asyncio
from fastapi import FastAPI, BackgroundTasks

DHCP_SERVER = "1.1.1.1"

# Set the event loop policy to WindowsProactorEventLoopPolicy
if asyncio.get_event_loop().is_closed():
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
else:
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())

app = FastAPI()

@app.get("/")
async def subprocess_test(background_tasks: BackgroundTasks):
    background_tasks.add_task(run_subprocess)
    return {"message": "Background task started"}

async def run_subprocess():
    proc = await asyncio.create_subprocess_exec(
        'powershell.exe',
        f'Get-Dhcp-Serverv4Scope -ComputerName \"{DHCP_SERVER}\"',
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )

    stdout, stderr = await proc.communicate()

    if stderr:
        print(stderr.decode())
    else:
        print(stdout.decode())

Explanation:

  1. Set Event Loop Policy:

    if asyncio.get_event_loop().is_closed():
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
else:
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())

    This code sets the event loop policy to asyncio.WindowsProactorEventLoopPolicy at the start of your script. This policy supports subprocesses on Windows.

  2. Decode Output:

    if stderr:
    print(stderr.decode())
else:
    print(stdout.decode())

    Decoding the output from stdout and stderr ensures that you get readable strings instead of bytes.

With these changes, your run_subprocess function should work correctly within the FastAPI background task on Windows.

#fastapi