Pyjo.Steps - Steps of a coroutine, blocking or non-blocking

from Pyjo.Steps import Command, Request, Sleep, Steps
from Pyjo.UserAgent import UserAgent

ua = UserAgent()

async def hello():
    tx = await Request(ua.build_tx('GET', 'https://mojolicious.org'))
    await Sleep(0.5)
    code, out, err = await Command(['hostname'])
    return f"{out.decode().strip()}: {tx.result.dom().at('title').text}"

print(Steps(hello()).run(ua))
Steps(hello()).run_p(ua).then(lambda text: print(text)).wait()

# Or import the module, as in Google Python Style Guide

import Pyjo.Steps

# The steps of another coroutine, awaited
async def twice():
    return [await Pyjo.Steps.Steps(hello()), await Pyjo.Steps.Steps(hello())]

Pyjo.Steps runs a coroutine that awaits transactions of HTTP, programs and sleeps, blocking and non-blocking, so that the code that makes requests and runs programs is written once, for both, as a library that a script and an action of a server share:

• a Request gives the transaction back with its response, or raises the Pyjo.Exception of an error of the connection;

• a Command gives the exit code, the output and the error of the program, or raises the Pyjo.Exception of a program that cannot be run;

• a Sleep waits for a number of seconds;

• Steps of another coroutine give what it returns.

A blocking request of Pyjo.UserAgent cannot run the event loop that is running already, as in an action of a server, and a non-blocking one cannot be waited for there either. The same steps run with run() in a script and with run_p() in the event loop.

No event loop runs the coroutine, so it awaits nothing else, not a Pyjo.Promise and not a future of asyncio, which is a TypeError. A generator that yields the requests, the programs and the sleeps, and takes Steps with yield from, runs as a coroutine does.

See also

Pyjo.UserAgent, Pyjo.IOLoop.ChildProcess, Pyjo.Promise.

Classes

class Pyjo.Steps.Command(args: Sequence[str], input: bytes = b'', *, env: Mapping[str, str] | None = None, timeout: float | None = None)

A program to run, with its arguments, what it gets on its standard input, its environment, the one of the process by default, and the number of seconds that it may run, without a limit by default.

code, out, err = await Command.new(['hostname'])
code, out, err = await Command.new(['/opt/bin/token'], env={**os.environ, 'AUDIENCE': audience}, timeout=30)

__await__() → Generator[Self, Any, tuple[int | None, bytes, bytes]]

code, out, err = await command

Run the program and give its exit code, its standard output and its standard error, or an exception of a program that runs longer than its timeout, which is killed.

class Pyjo.Steps.Pyjo_Steps(steps: Coroutine[Any, Any, T] | Generator[Operation, Any, T])

Pyjo.Steps inherits all attributes and methods from Pyjo.Base and implements the following new ones. The steps are of one coroutine, which they run once.

__await__() → Generator[Operation, Any, T]

value = await steps

Wait for the steps in another coroutine and give what they return.

__iter__() → Generator[Operation, Any, T]

value = yield from steps

Same as __await__(), in a generator.

run(ua: Pyjo_UserAgent) → T

value = steps.run(ua)

Run the steps blocking, the requests with a user agent, of Pyjo.UserAgent.Pyjo_UserAgent.start(), the programs of subprocess.run() and the sleeps of time.sleep(), and give what they return.

run_p(ua: Pyjo_UserAgent) → Pyjo_Promise

promise = steps.run_p(ua)

Same as run(), but non-blocking, with Pyjo.UserAgent.Pyjo_UserAgent.start_p(), Pyjo.IOLoop.ChildProcess.Pyjo_IOLoop_ChildProcess.run_p() and Pyjo.Promise.Pyjo_Promise.timer(), and returns a Pyjo.Promise object.

class Pyjo.Steps.Request(tx: Pyjo_Transaction_HTTP, ua: Pyjo_UserAgent | None = None)

A transaction to start, with the user agent that starts it, or the one of run() by default, as one whose event start adds the credentials of a client to it.

tx = await Request.new(ua.build_tx('GET', 'https://mojolicious.org'))
tx = await Request.new(tx, ua)

__await__() → Generator[Self, Any, Pyjo_Transaction_HTTP]

tx = await request

Start the transaction and give it back with its response.

class Pyjo.Steps.Sleep(seconds: float)

A number of seconds to wait, as for a file that another program writes, or before a request is tried again.

await Sleep.new(0.5)

__await__() → Generator[Self, Any, None]

await sleep

Wait for the seconds, with time.sleep() of run() and with a timer of the reactor of run_p(), which runs the other events meanwhile.