Skip to content

FastAgency

fastagency.FastAgency #

FastAgency(
    provider: ProviderProtocol,
    ui: UIBase,
    *,
    title: Optional[str] = None,
    description: Optional[str] = None
)

Initialize the FastAgency object.

PARAMETER DESCRIPTION
provider

The provider object to use

TYPE: ProviderProtocol

ui

The UI object to use

TYPE: UI

title

The title of the FastAgency. If None, the default string will be used. Defaults to None.

TYPE: Optional[str] DEFAULT: None

description

The description of the FastAgency. If None, the default string will be used. Defaults to None.

TYPE: Optional[str] DEFAULT: None

Source code in fastagency/app.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def __init__(
    self,
    provider: ProviderProtocol,
    ui: UIBase,
    *,
    title: Optional[str] = None,
    description: Optional[str] = None,
) -> None:
    """Initialize the FastAgency object.

    Args:
        provider (ProviderProtocol): The provider object to use
        ui (UI): The UI object to use
        title (Optional[str], optional): The title of the FastAgency. If None, the default string will be used. Defaults to None.
        description (Optional[str], optional): The description of the FastAgency. If None, the default string will be used. Defaults to None.
    """
    # check if we need to start coverage
    logger.info("Checking if coverage is needed.")
    coverage_process_start = os.environ.get("COVERAGE_PROCESS_START")
    if coverage_process_start:
        logger.info("Coverage process start detected")
        logger.info(f"Coverage configuration file: {coverage_process_start}")
        logger.info(
            "To ensure coverage is written out, terminate this program with SIGTERM"
        )
        import coverage

        coverage.process_startup()
    _self: Runnable = self
    self._title = title or "FastAgency application"
    self._description = description or "FastAgency application"

    logger.info(
        f"Initializing FastAgency {self} with workflows: {provider} and UI: {ui}"
    )
    self._provider = provider
    self._ui = ui
    logger.info(f"Initialized FastAgency: {self}")

description property #

description: str

Return the description of the FastAgency.

provider property #

provider: ProviderProtocol

Return the provider object.

title property #

title: str

Return the title of the FastAgency.

ui property #

ui: UIBase

Return the UI object.

create #

create(import_string: str) -> Generator[None, None, None]

Create the FastAgency.

Source code in fastagency/app.py 
88
89
90
91
92
@contextmanager
def create(self, import_string: str) -> Generator[None, None, None]:
    """Create the FastAgency."""
    with self._ui.create(app=self, import_string=import_string):
        yield

handle_asgi async #

handle_asgi(
    scope: dict[str, Any],
    receive: Callable[[dict[str, Any]], Awaitable[None]],
    send: Callable[[dict[str, Any]], Awaitable[None]],
) -> None
Source code in fastagency/app.py 
140
141
142
143
144
145
146
147
148
149
150
151
async def handle_asgi(
    self,
    scope: dict[str, Any],
    receive: Callable[[dict[str, Any]], Awaitable[None]],
    send: Callable[[dict[str, Any]], Awaitable[None]],
) -> None:
    if isinstance(self.ui, ASGIProtocol):
        return await self.ui.handle_asgi(self, scope, receive, send)
    else:
        raise FastAgencyASGINotImplementedError(
            f"ASGI interface not supported for UI: {self.ui.__class__.__name__}. Try running with 'fastapi run' or with a WSGI server like 'gunicorn'/'waitress'."
        )

handle_wsgi #

handle_wsgi(
    environ: dict[str, Any],
    start_response: Callable[..., Any],
) -> list[bytes]
Source code in fastagency/app.py 
129
130
131
132
133
134
135
136
137
138
def handle_wsgi(
    self, environ: dict[str, Any], start_response: Callable[..., Any]
) -> list[bytes]:
    logger.debug(f"Handling WSGI request: {environ}")
    if isinstance(self.ui, WSGIProtocol):
        return self.ui.handle_wsgi(self, environ, start_response)
    else:
        raise FastAgencyWSGINotImplementedError(
            f"WSGI interface not supported for UI: {self.ui.__class__.__name__}. Try running with 'fastapi run' or with a ASGI server like 'uvicorn'."
        )

start #

start(
    *,
    import_string: str,
    name: Optional[str] = None,
    params: dict[str, Any],
    single_run: bool = False
) -> None

Start the FastAgency.

Source code in fastagency/app.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def start(
    self,
    *,
    import_string: str,
    name: Optional[str] = None,
    params: dict[str, Any],
    single_run: bool = False,
) -> None:
    """Start the FastAgency."""
    self.ui.start(
        app=self,
        import_string=import_string,
        name=name,
        params=params,
        single_run=single_run,
    )