chatsky_ui.api.api_v1.endpoints package

chatsky_ui.api.api_v1.endpoints.bot module

async chatsky_ui.api.api_v1.endpoints.bot.check_build_processes(build_id: int | None = None, build_manager: BuildManager = Depends(get_build_manager), run_manager: RunManager = Depends(get_run_manager), pagination: Pagination = Depends(NoneType)) Dict[str, Any] | List[Dict[str, Any]] | None[source]

Checks the status of all build processes and returns them along with their runs info.

The offset and limit parameters can be used to paginate the results.

Args:

build_id (Optional[int]): The id of the process to check. If not specified, all processes will be returned.

async chatsky_ui.api.api_v1.endpoints.bot.check_build_status(*, build_id: int, build_manager: BuildManager = Depends(get_build_manager)) Dict[str, str][source]

Checks the status of a build process with the given id.

Args:

build_id (int): The id of the process to check. build_manager (BuildManager): The process manager dependency to check the process with.

Raises:

HTTPException: With status code 404 if the process is not found.

Returns:

{“status”: “completed”}: in case of a successfully completed process. {“status”: “running”}: in case of a still running process. {“status”: “stopped”}: in case of a stopped process. {“status”: “failed”}: in case of a failed-to-run process.

async chatsky_ui.api.api_v1.endpoints.bot.check_run_processes(run_id: int | None = None, run_manager: RunManager = Depends(get_run_manager), pagination: Pagination = Depends(NoneType)) Dict[str, Any] | List[Dict[str, Any]] | None[source]

Checks the status of all run processes and returns them.

The offset and limit parameters can be used to paginate the results.

Args:

run_id (Optional[int]): The id of the process to check. If not specified, all processes will be returned.

async chatsky_ui.api.api_v1.endpoints.bot.check_run_status(*, run_id: int, run_manager: RunManager = Depends(get_run_manager)) Dict[str, Any][source]

Checks the status of a run process with the given id.

Args:

build_id (int): The id of the process to check. run_manager (RunManager): The process manager dependency to check the process with.

Raises:

HTTPException: With status code 404 if the process is not found.

Returns:

{“status”: “alive”}: in case of a successfully run process. Now it is able to communicate. {“status”: “running”}: in case of a still running process. {“status”: “stopped”}: in case of a stopped process. {“status”: “failed”}: in case of a failed-to-run process.

async chatsky_ui.api.api_v1.endpoints.bot.connect(websocket: WebSocket, websocket_manager: WebSocketManager = Depends(get_websocket_manager), run_manager: RunManager = Depends(get_run_manager)) None[source]

Establishes a WebSocket connection to communicate with an alive run process identified by its ‘run_id’.

The WebSocket URL should adhere to the format: /bot/run/connect?run_id=<run_id>.

async chatsky_ui.api.api_v1.endpoints.bot.get_build_logs(build_id: int, build_manager: BuildManager = Depends(get_build_manager), pagination: Pagination = Depends(NoneType)) List[str] | None[source]

Gets the logs of a specific build process.

The offset and limit parameters can be used to paginate the results.

async chatsky_ui.api.api_v1.endpoints.bot.get_run_logs(run_id: int, run_manager: RunManager = Depends(get_run_manager), pagination: Pagination = Depends(NoneType)) List[str] | None[source]

Gets the logs of a specific run process.

The offset and limit parameters can be used to paginate the results.

async chatsky_ui.api.api_v1.endpoints.bot.start_build(preset: Preset, background_tasks: BackgroundTasks, build_manager: BuildManager = Depends(get_build_manager), index: Index = Depends(get_index)) Dict[str, str | int][source]

Starts a build process with the given preset.

This runs a background task to check the status of the process every 2 seconds.

Args:

preset (Preset): The preset to set the build process for. Must be among (“success”, “failure”, “loop”)

Returns:

{“status”: “ok”, “build_id”: build_id}: in case of starting the build process successfully.

async chatsky_ui.api.api_v1.endpoints.bot.start_run(*, build_id: int, preset: Preset, background_tasks: BackgroundTasks, run_manager: RunManager = Depends(get_run_manager)) Dict[str, str | int][source]

Starts a run process with the given preset.

This runs a background task to check the status of the process every 2 seconds.

Args:

build_id (int): The id of the build process to start running. preset (Preset): The preset to set the build process for. Must be among (“success”, “failure”, “loop”)

Returns:

{“status”: “ok”, “build_id”: run_id}: in case of starting the run process successfully.

async chatsky_ui.api.api_v1.endpoints.bot.stop_build(*, build_id: int, build_manager: BuildManager = Depends(get_build_manager)) Dict[str, str][source]

Stops a build process with the given id.

Args:

build_id (int): The id of the process to stop. build_id (BuildManager): The process manager dependency to stop the process with.

Raises:

HTTPException: With status code 404 if the process is not found.

Returns:

{“status”: “ok”}: in case of stopping a process successfully.

async chatsky_ui.api.api_v1.endpoints.bot.stop_run(*, run_id: int, run_manager: RunManager = Depends(get_run_manager)) Dict[str, str][source]

Stops a run process with the given id.

Args:

run_id (int): The id of the process to stop. run_manager (RunManager): The process manager dependency to stop the process with.

Raises:

HTTPException: With status code 404 if the process is not found.

Returns:

{“status”: “ok”}: in case of stopping a process successfully.

chatsky_ui.api.api_v1.endpoints.dff_services module

async chatsky_ui.api.api_v1.endpoints.dff_services.get_conditions() Dict[str, str | list][source]

Gets the dff’s out-of-the-box conditions.

async chatsky_ui.api.api_v1.endpoints.dff_services.lint_snippet(snippet: CodeSnippet) Dict[str, str][source]

Lints a snippet with Pylint.

This endpoint Joins the snippet with all imports existing in the conditions.py file and then runs Pylint on it.

async chatsky_ui.api.api_v1.endpoints.dff_services.search_service(service_name: str, index: Index = Depends(get_index)) Dict[str, str | list | None][source]

Searches for a custom service by name and returns its code.

A service could be a condition, reponse, or pre/postservice.

chatsky_ui.api.api_v1.endpoints.flows module

async chatsky_ui.api.api_v1.endpoints.flows.flows_get() Dict[str, str | Dict[str, list]][source]

Get the flows by reading the frontend_flows.yaml file.

async chatsky_ui.api.api_v1.endpoints.flows.flows_post(flows: Dict[str, list]) Dict[str, str][source]

Write the flows to the frontend_flows.yaml file.