Core API

Admin UI

Admin UI API router Provides a web-based dashboard for managing the Petal App Manager system.

async petal_app_manager.api.admin_ui.admin_dashboard(request: Request)

Admin dashboard with real-time log streaming

petal_app_manager.api.admin_ui.get_logger() → Logger

Get the logger instance.

Health Endpoints

async petal_app_manager.api.health.detailed_health_check() → HealthMessage

Comprehensive health check endpoint that reports the status of each proxy.

petal_app_manager.api.health.get_logger() → Logger | None

Get the logger instance.

async petal_app_manager.api.health.health_check() → BasicHealthResponse

Basic health check endpoint.

async petal_app_manager.api.health.organization_health_check() → OrganizationHealthResponse

Get current organization information and status.

Configuration API

class petal_app_manager.api.config_api.AllComponentsResponse(*, petals: List[PetalInfo], proxies: List[ProxyInfo], total_petals: int, total_proxies: int)

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

petals: List[PetalInfo]

proxies: List[ProxyInfo]

total_petals: int

total_proxies: int

class petal_app_manager.api.config_api.ConfigResponse(*, enabled_proxies: List[str], enabled_petals: List[str], petal_dependencies: Dict[str, List[str]], proxy_dependencies: Dict[str, List[str]])

Bases: BaseModel

enabled_petals: List[str]

enabled_proxies: List[str]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

petal_dependencies: Dict[str, List[str]]

proxy_dependencies: Dict[str, List[str]]

class petal_app_manager.api.config_api.PetalControlRequest(*, petals: List[str], action: str)

Bases: BaseModel

action: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

petals: List[str]

class petal_app_manager.api.config_api.PetalInfo(*, name: str, enabled: bool, dependencies: List[str])

Bases: BaseModel

dependencies: List[str]

enabled: bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

class petal_app_manager.api.config_api.ProxyInfo(*, name: str, enabled: bool, dependencies: List[str], dependents: List[str])

Bases: BaseModel

dependencies: List[str]

dependents: List[str]

enabled: bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

async petal_app_manager.api.config_api.control_petals(request: PetalControlRequest) → Dict[str, Any]

Enable or disable one or more petals

async petal_app_manager.api.config_api.control_proxies(request: PetalControlRequest) → Dict[str, Any]

Enable or disable one or more proxies

petal_app_manager.api.config_api.get_logger() → Logger

Get the logger instance.

async petal_app_manager.api.config_api.get_status() → ConfigResponse

Get current configuration status

async petal_app_manager.api.config_api.list_all_components()

List all available petals and proxies, regardless of their enabled/disabled state

Cloud API

class petal_app_manager.api.cloud_api.GetItemRequest(*, table_name: str, key_name: str, key_value: str)

Bases: BaseModel

key_name: str

key_value: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

table_name: str

class petal_app_manager.api.cloud_api.ScanTableRequest(*, table_name: str, filters: list = [])

Bases: BaseModel

filters: list

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

table_name: str

class petal_app_manager.api.cloud_api.SetItemRequest(*, table_name: str, item_data: Dict[str, Any])

Bases: BaseModel

item_data: Dict[str, Any]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

table_name: str

class petal_app_manager.api.cloud_api.UpdateItemRequest(*, table_name: str, key_name: str, key_value: str, update_data: Dict[str, Any])

Bases: BaseModel

key_name: str

key_value: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

table_name: str

update_data: Dict[str, Any]

async petal_app_manager.api.cloud_api.delete_item(request: GetItemRequest) → Dict[str, Any]

Delete an item from the cloud database.

async petal_app_manager.api.cloud_api.get_item(request: GetItemRequest) → Dict[str, Any]

Get a specific item from the cloud database.

petal_app_manager.api.cloud_api.get_logger() → Logger | None

Get the logger instance.

async petal_app_manager.api.cloud_api.scan_table(request: ScanTableRequest) → Dict[str, Any]

List all data in a particular table in the cloud database.

async petal_app_manager.api.cloud_api.set_item(request: SetItemRequest) → Dict[str, Any]

Create or overwrite an item in the cloud database.

async petal_app_manager.api.cloud_api.update_item(request: UpdateItemRequest) → Dict[str, Any]

Update an existing item in the cloud database.

Bucket API

async petal_app_manager.api.bucket_api.delete_file_test(s3_key: str) → Dict[str, Any]

Delete a flight log file from S3 bucket for testing.

async petal_app_manager.api.bucket_api.download_file_test(s3_key: str, return_file: bool = Query(False)) → Dict[str, Any]

Download a flight log file from S3 bucket for testing.

async petal_app_manager.api.bucket_api.get_info() → Dict[str, Any]

Get S3 bucket proxy information.

petal_app_manager.api.bucket_api.get_logger() → Logger

Get the logger instance.

async petal_app_manager.api.bucket_api.list_files_test(prefix: str | None = Query(None), max_keys: int = Query(100)) → Dict[str, Any]

List flight log files in S3 bucket for testing.

async petal_app_manager.api.bucket_api.upload_file_test(file: UploadFile = File(PydanticUndefined), custom_filename: str | None = Form(None)) → Dict[str, Any]

Upload a flight log file to S3 bucket storage for testing.

MAVLink FTP API

class petal_app_manager.api.mavftp_api.ClearFailLogsRequest(*, remote_path: str = '/fs/microsd')

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

remote_path: str

async petal_app_manager.api.mavftp_api.clear_fail_logs(request: ClearFailLogsRequest) → Dict[str, Any]

Clear all fail_*.log files from the vehicle’s filesystem.

petal_app_manager.api.mavftp_api.get_logger() → Logger | None

Get the logger instance.

async petal_app_manager.api.mavftp_api.list_directory(request: ClearFailLogsRequest) → Dict[str, Any]

Clear all fail_*.log files from the vehicle’s filesystem.

MQTT API

class petal_app_manager.api.mqtt_api.MQTTResponse(*, status: str, message: str | None = None)

Bases: BaseModel

message: str | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

status: str

class petal_app_manager.api.mqtt_api.PublishMessageRequest(*, topic: str, payload: Dict[str, Any], qos: int = 1)

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

payload: Dict[str, Any]

qos: int

topic: str

class petal_app_manager.api.mqtt_api.SubscribePatternRequest(*, pattern: str)

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pattern: str

class petal_app_manager.api.mqtt_api.SubscribeTopicRequest(*, topic: str)

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

topic: str

class petal_app_manager.api.mqtt_api.UnsubscribeTopicRequest(*, topic: str)

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

topic: str

petal_app_manager.api.mqtt_api.get_logger() → Logger | None

Get the logger instance.

async petal_app_manager.api.mqtt_api.get_mqtt_status() → Dict[str, Any]

Get MQTT proxy status and health information.

async petal_app_manager.api.mqtt_api.list_subscriptions() → Dict[str, Any]

List current MQTT subscriptions.

async petal_app_manager.api.mqtt_api.publish_message(request: PublishMessageRequest) → MQTTResponse

Publish a message to an MQTT topic.

async petal_app_manager.api.mqtt_api.subscribe_topic(request: SubscribeTopicRequest) → MQTTResponse

Subscribe to an MQTT topic.

async petal_app_manager.api.mqtt_api.unsubscribe_topic(request: UnsubscribeTopicRequest) → MQTTResponse

Unsubscribe from an MQTT topic.

Proxy Info

petal_app_manager.api.proxy_info.get_logger() → Logger | None

Get the logger instance.

async petal_app_manager.api.proxy_info.list_proxies()

List all available proxies in the system.