// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
using System.Text;
using System.Globalization;
namespace Aspire.Hosting.CodeGeneration.Python;
/// <summary>
/// Represents a builder for generating Python module structure and related code components.
/// Sections are built independently and combined in the correct order when Write() is called.
/// </summary>
internal sealed class PythonModuleBuilder
{
/// <summary>
/// Gets the enum type definitions.
/// </summary>
public StringBuilder Enums { get; } = new();
/// <summary>
/// Gets the DTO class definitions.
/// </summary>
public StringBuilder DtoClasses { get; } = new();
/// <summary>
/// Gets the exported value definitions.
/// </summary>
public StringBuilder ExportedValues { get; } = new();
/// <summary>
/// Gets the type class definitions (context types, wrapper types).
/// </summary>
public Dictionary<string, StringBuilder> TypeClasses { get; } = new();
/// <summary>
/// Gets the interface class definitions
/// </summary>
public Dictionary<string, StringBuilder> InterfaceClasses { get; } = new();
/// <summary>
/// Gets the resource class builder definitions.
/// </summary>
public Dictionary<string, StringBuilder> ResourceBuilders { get; } = new();
/// <summary>
/// Gets the resource parameter definitions
/// </summary>
public Dictionary<string, StringBuilder> ResourceOptions { get; } = new();
/// <summary>
/// A list of all keyword argument names supported for a Resource builder, used to prevent naming conflicts.
/// </summary>
public Dictionary<string, List<string>> ResourceOptionNames { get; } = new();
/// <summary>
/// Gets the resource class definitions
/// </summary>
public Dictionary<string, StringBuilder> ResourceClasses { get; } = new();
/// <summary>
/// Gets the entry point function definitions.
/// </summary>
public StringBuilder EntryPoints { get; } = new();
/// <summary>
/// Gets the handle registration definitions.
/// </summary>
public Dictionary<string, string> HandleRegistrations { get; } = new();
/// <summary>
/// Gets the method parameter type class definitions, keyed by parameter type name.
/// </summary>
public Dictionary<string, StringBuilder> MethodParameters { get; } = new();
/// <summary>
/// Writes the complete Python module content.
/// </summary>
/// <returns>The complete Python module as a string.</returns>
public string Write()
{
var output = new StringBuilder();
output.AppendLine(Header);
output.AppendLine(StandardImports);
// TODO: Not quite sure how to handle versioning, as users wont have direct control over the
// code generator version. Ideally this would reflect the .NET package version, but that will be
// painful for test snapshots, just going to hardcode a version for now.
output.AppendLine(CultureInfo.InvariantCulture, $"__version__ = \"0.1.0\"");
output.AppendLine();
output.AppendLine(Utils);
// Enums
if (Enums.Length > 0)
{
output.AppendLine();
output.AppendLine("# ============================================================================");
output.AppendLine("# Enum Types");
output.AppendLine("# ============================================================================");
output.AppendLine();
output.Append(Enums);
}
// Method parameters
if (MethodParameters.Count > 0)
{
output.AppendLine();
output.AppendLine("# ============================================================================");
output.AppendLine("# Method Parameters");
output.AppendLine("# ============================================================================");
foreach (var kvp in MethodParameters)
{
output.AppendLine();
output.Append(kvp.Value);
}
}
// DTO Classes
if (DtoClasses.Length > 0)
{
output.AppendLine();
output.AppendLine("# ============================================================================");
output.AppendLine("# DTO Classes (Data Transfer Objects)");
output.AppendLine("# ============================================================================");
output.AppendLine();
output.Append(DtoClasses);
}
if (ExportedValues.Length > 0)
{
output.AppendLine();
output.AppendLine("# ============================================================================");
output.AppendLine("# Exported Values");
output.AppendLine("# ============================================================================");
output.AppendLine();
output.Append(ExportedValues);
}
// Type Classes
if (TypeClasses.Count > 0)
{
output.AppendLine();
output.AppendLine("# ============================================================================");
output.AppendLine("# Type Classes");
output.AppendLine("# ============================================================================");
foreach (var kvp in TypeClasses)
{
output.AppendLine();
output.Append(kvp.Value);
}
}
// Interface Classes
if (InterfaceClasses.Count > 0)
{
output.AppendLine();
output.AppendLine("# ============================================================================");
output.AppendLine("# Interface Classes");
output.AppendLine("# ============================================================================");
foreach (var kvp in InterfaceClasses)
{
output.AppendLine();
output.Append(kvp.Value);
}
}
// Resource Builder Classes
if (ResourceClasses.Count > 0)
{
output.AppendLine();
output.AppendLine("# ============================================================================");
output.AppendLine("# Builder Classes");
output.AppendLine("# ============================================================================");
foreach (var kvp in ResourceClasses)
{
output.AppendLine();
output.Append(ResourceOptions[kvp.Key]);
output.AppendLine();
output.Append(kvp.Value);
}
}
// Entry Points
if (EntryPoints.Length > 0)
{
output.AppendLine();
output.AppendLine("# ============================================================================");
output.AppendLine("# Entry Point Functions");
output.AppendLine("# ============================================================================");
output.AppendLine();
output.Append(EntryPoints);
}
// Connection Helper
output.AppendLine();
output.AppendLine("# ============================================================================");
output.AppendLine("# Connection Helper");
output.AppendLine("# ============================================================================");
output.AppendLine();
output.AppendLine(ConnectionHelperCode);
// Handle Registrations
if (HandleRegistrations.Count > 0)
{
output.AppendLine();
output.AppendLine();
output.AppendLine("# ============================================================================");
output.AppendLine("# Handle Registrations");
output.AppendLine("# ============================================================================");
output.AppendLine();
output.AppendLine("_register_handle_wrapper(\"Aspire.Hosting/Aspire.Hosting.ApplicationModel.ReferenceExpression\", lambda handle, _: ReferenceExpression(handle))");
output.AppendLine("_register_handle_wrapper(\"System.Private.CoreLib/System.Threading.CancellationToken\", CancellationToken)");
foreach (var (typeId, className) in HandleRegistrations)
{
output.AppendLine(
CultureInfo.InvariantCulture,
$"_register_handle_wrapper(\"{typeId}\", {className})");
}
}
return output.ToString();
}
/// <summary>
/// The file header with copyright notice.
/// </summary>
public const string Header = """
# -------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See LICENSE in project root for information.
#
# This is a generated file. Any modifications may be overwritten.
# -------------------------------------------------------------
""";
/// <summary>
/// Standard Python imports for the generated SDK.
/// </summary>
public const string StandardImports = """
from __future__ import annotations
import os
import base64
import sys
import json
import logging
import secrets
import signal
import socket
import threading
import time
import abc
import datetime
import types
import typing
from functools import cached_property as _cached_property
from contextlib import AbstractContextManager
_logger = logging.getLogger(__name__)
# Maximum allowed message size (64 MB) to prevent memory exhaustion from malicious Content-Length
_MAX_MESSAGE_SIZE = 64 * 1024 * 1024
# Maximum number of headers and total header size to prevent header-flooding attacks
_MAX_HEADER_COUNT = 16
_MAX_HEADER_BYTES = 8 * 1024
# Marker string for detecting generic .NET builder type names.
_BUILDER_GENERIC_MARKER = "Builder`1["
_uncached_property = property
""";
/// <summary>
/// Utility functions for the generated SDK.
/// </summary>
public const string Utils = """
# ============================================================================
# Platform-specific Socket Implementation
# ============================================================================
if sys.platform == "win32":
import ctypes
from ctypes import wintypes
_kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
class _OVERLAPPED(ctypes.Structure):
'''Windows OVERLAPPED structure for async I/O.'''
_fields_ = [
("Internal", ctypes.POINTER(ctypes.c_ulong)),
("InternalHigh", ctypes.POINTER(ctypes.c_ulong)),
("Offset", wintypes.DWORD),
("OffsetHigh", wintypes.DWORD),
("hEvent", wintypes.HANDLE),
]
class _PipeSocket:
'''A socket-like wrapper around a Win32 named pipe using ctypes.'''
# Win32 constants
GENERIC_READ = 0x80000000
GENERIC_WRITE = 0x40000000
OPEN_EXISTING = 3
INVALID_HANDLE_VALUE = -1
FILE_FLAG_OVERLAPPED = 0x40000000
ERROR_IO_PENDING = 997
ERROR_FILE_NOT_FOUND = 2
def __init__(self, pipe_path: str) -> None:
self._handle: int | None = None
handle = _kernel32.CreateFileW(
pipe_path,
self.GENERIC_READ | self.GENERIC_WRITE,
0, # no sharing
None, # default security
self.OPEN_EXISTING,
self.FILE_FLAG_OVERLAPPED, # match server's async mode
None # no template
)
if handle == self.INVALID_HANDLE_VALUE:
error = ctypes.get_last_error()
if error == self.ERROR_FILE_NOT_FOUND:
raise FileNotFoundError(f"Pipe not found: {pipe_path}")
raise OSError(f"CreateFile failed with error {error}")
self._handle = handle
def _create_overlapped_event(self) -> _OVERLAPPED:
'''Create an OVERLAPPED structure with an event for async I/O.'''
overlapped = _OVERLAPPED()
overlapped.hEvent = _kernel32.CreateEventW(None, True, False, None)
return overlapped
def recv(self, n: int) -> bytes:
'''Read up to n bytes using overlapped I/O.'''
buffer = ctypes.create_string_buffer(n)
bytes_read = wintypes.DWORD()
overlapped = self._create_overlapped_event()
try:
success = _kernel32.ReadFile(
self._handle,
buffer,
n,
ctypes.byref(bytes_read),
ctypes.byref(overlapped)
)
if not success:
error = ctypes.get_last_error()
if error == self.ERROR_IO_PENDING:
# Wait for the operation to complete
_kernel32.GetOverlappedResult(
self._handle,
ctypes.byref(overlapped),
ctypes.byref(bytes_read),
True # wait
)
else:
raise OSError(f"ReadFile failed with error {error}")
finally:
_kernel32.CloseHandle(overlapped.hEvent)
return buffer.raw[:bytes_read.value]
def sendall(self, data: bytes) -> None:
'''Write all data using overlapped I/O.'''
bytes_written = wintypes.DWORD()
overlapped = self._create_overlapped_event()
try:
offset = 0
while offset < len(data):
chunk = data[offset:]
success = _kernel32.WriteFile(
self._handle,
chunk,
len(chunk),
ctypes.byref(bytes_written),
ctypes.byref(overlapped)
)
if not success:
error = ctypes.get_last_error()
if error == self.ERROR_IO_PENDING:
# Wait for the operation to complete
_kernel32.GetOverlappedResult(
self._handle,
ctypes.byref(overlapped),
ctypes.byref(bytes_written),
True # wait
)
else:
raise OSError(f"WriteFile failed with error {error}")
offset += bytes_written.value
finally:
_kernel32.CloseHandle(overlapped.hEvent)
def close(self) -> None:
'''Close the handle.'''
if self._handle is not None and self._handle != self.INVALID_HANDLE_VALUE:
_kernel32.CloseHandle(self._handle)
self._handle = None
def _connect_pipe(socket_path: str, timeout_sec: float) -> _PipeSocket:
'''Connect to a named pipe with timeout, retrying until available.'''
pipe_path = f"\\\\.\\pipe\\{socket_path}"
_logger.debug("Connecting to: %s", pipe_path)
start_time = time.time()
while (time.time() - start_time) < timeout_sec:
try:
return _PipeSocket(pipe_path)
except FileNotFoundError:
time.sleep(0.1)
raise TimeoutError("Connection timeout")
else:
# On Unix, use socket.socket directly as the pipe socket type
_PipeSocket = socket.socket # type: ignore[misc]
def _connect_pipe(socket_path: str, timeout_sec: float) -> _PipeSocket:
'''Connect to a Unix domain socket with timeout.'''
_logger.debug("Connecting to: %s", socket_path)
start_time = time.time()
while (time.time() - start_time) < timeout_sec:
try:
sock = _PipeSocket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.settimeout(timeout_sec)
sock.connect(socket_path)
sock.settimeout(None) # Set to blocking mode
return sock
except FileNotFoundError:
time.sleep(0.1)
raise TimeoutError("Connection timeout")
# ============================================================================
# Base Types
# ============================================================================
class AtsErrorDetails(typing.TypedDict, total=False):
'''Error details for ATS errors.'''
parameter: str
expected: str
actual: str
class AtsErrorData(typing.TypedDict, total=False):
'''Structured error from ATS capability invocation.'''
code: str
message: str
capability: str
details: AtsErrorDetails
class AtsErrorCodes:
'''ATS error codes returned by the server.'''
CAPABILITY_NOT_FOUND = "CAPABILITY_NOT_FOUND"
HANDLE_NOT_FOUND = "HANDLE_NOT_FOUND"
TYPE_MISMATCH = "TYPE_MISMATCH"
INVALID_ARGUMENT = "INVALID_ARGUMENT"
ARGUMENT_OUT_OF_RANGE = "ARGUMENT_OUT_OF_RANGE"
CALLBACK_ERROR = "CALLBACK_ERROR"
INTERNAL_ERROR = "INTERNAL_ERROR"
def _simplify_type_name(typestring: str) -> str:
'''Simplify a .NET type name for readability in error messages.'''
# Collapse any embedded newlines/whitespace so patterns can match
collapsed = " ".join(typestring.split())
# Check for generic builder pattern like "IResourceBuilder`1[Namespace.Type]"
idx = collapsed.find(_BUILDER_GENERIC_MARKER)
if idx >= 0:
inner = collapsed[idx + len(_BUILDER_GENERIC_MARKER):]
# Skip any extra opening brackets (e.g. "[[")
inner = inner.lstrip("[").strip()
# Take only the qualified name (before any comma/assembly info or closing brackets)
inner = inner.split(",")[0].split("]")[0].strip()
# Get simple class name from fully-qualified name
inner_type = inner.rsplit(".", 1)[-1]
# Strip leading 'I' from interface names (e.g., IResourceWithConnectionString -> ResourceWithConnectionString)
if len(inner_type) > 1 and inner_type[0] == "I" and inner_type[1].isupper():
inner_type = inner_type[1:]
return inner_type
return typestring.split(".")[-1] # Fallback: take simple name from fully-qualified type
def _format_type_error(error: AtsErrorData) -> Exception:
'''Formats an ATS type error into a Python exception with a helpful message.'''
try:
details = error["details"] # type: ignore[index]
parameter = details["parameter"] # type: ignore[index]
expected = _simplify_type_name(details["expected"]) # type: ignore[index]
actual = _simplify_type_name(details["actual"]) # type: ignore[index]
message_prefix = ""
if capability := error.get("capability"):
message_prefix = f" in '{capability.split('/')[-1]}'"
message = f"Type mismatch{message_prefix} for parameter '{parameter}': expected {expected}, got {actual}"
return TypeError(message)
except KeyError:
if message := error.get("message"):
return TypeError(message)
if capability := error.get("capability"):
return TypeError(f"Type mismatch in capability '{capability}'")
return TypeError("Parameter type mismatch.")
def _is_ats_error(value: typing.Any) -> bool:
'''Type guard to check if a value is an ATS error response.'''
return (
isinstance(value, dict)
and "$error" in value
and isinstance(value["$error"], dict)
)
def _is_marshalled_handle(value: typing.Any) -> bool:
'''Type guard to check if a value is a marshalled handle.'''
return (
isinstance(value, dict)
and "$handle" in value
and "$type" in value
)
# ============================================================================
# Handle
# ============================================================================
class Handle:
'''
A typed handle to a .NET object in the ATS system.
Handles are opaque references that can be passed to capabilities.
'''
def __init__(self, handle_data: typing.Mapping[str, typing.Any]) -> None:
self._handle_id = handle_data["$handle"]
self._type_id = handle_data["$type"]
@property
def handle_id(self) -> str:
'''The handle ID (instance number)'''
return self._handle_id
@property
def type_id(self) -> str:
'''The ATS type ID'''
return self._type_id
def to_json(self) -> typing.Mapping[str, typing.Any]:
'''Serialize for JSON-RPC transport'''
return {
"$handle": self._handle_id,
"$type": self._type_id
}
def __repr__(self) -> str:
'''String representation for debugging'''
friendly_name = self._type_id.split("/")[-1].split(".")[-1]
return f"Handle[{friendly_name}]({self._handle_id})"
# ============================================================================
# Handle Wrapper Registry
# ============================================================================
_HandleWrapperFactory = typing.Callable[[Handle, "AspireClient"], typing.Any]
# Registry of handle wrapper factories by type ID
_handle_wrapper_registry: dict[str, _HandleWrapperFactory] = {}
def _register_handle_wrapper(type_id: str, factory: _HandleWrapperFactory) -> None:
'''
Register a wrapper factory for a type ID.
Called by generated code to register wrapper classes.
'''
_handle_wrapper_registry[type_id] = factory
def _wrap_if_handle(value: typing.Any, client: AspireClient | None = None, kwargs: typing.Mapping[str, typing.Any] | None = None) -> typing.Any:
'''
Checks if a value is a marshalled handle and wraps it appropriately.
Uses the wrapper registry to create typed wrapper instances when available.
'''
if isinstance(value, dict) and _is_marshalled_handle(value):
handle = Handle(value)
type_id = value["$type"]
# Try to find a registered wrapper factory for this type
if type_id and client:
factory = _handle_wrapper_registry.get(type_id)
if factory:
if kwargs:
return factory(handle, client, **kwargs)
return factory(handle, client)
return handle
return value
# ============================================================================
# Errors
# ============================================================================
class AspireError(Exception):
'''Base class for Aspire errors.'''
error: AtsErrorData
def __init__(self, error: AtsErrorData) -> None:
super().__init__(error.get("message", "Unknown error"))
self.error = error
@property
def code(self) -> str:
'''Machine-readable error code'''
return self.error.get("code", "UNKNOWN")
@property
def capability(self) -> str | None:
'''The capability that failed (if applicable)'''
return self.error.get("capability")
class OperationCancelled(Exception):
'''Error thrown when a cancellation token is invoked.'''
pass
# ============================================================================
# JSON Encoder
# ============================================================================
@typing.runtime_checkable
class _ReferenceHandle(typing.Protocol):
'''Protocol for objects that have a handle property.'''
@property
def handle(self) -> Handle:
...
def _timedelta_as_isostr(td: datetime.timedelta) -> str:
'''Converts a datetime.timedelta object into an ISO 8601 formatted string, e.g. 'P4DT12H30M05S'
Function adapted from the Tin Can Python project: https://github.com/RusticiSoftware/TinCanPython
:param td: The timedelta object to convert
:type td: datetime.timedelta
:return: An ISO 8601 formatted string representing the timedelta object
:rtype: str
'''
# Split seconds to larger units
seconds = td.total_seconds()
minutes, seconds = divmod(seconds, 60)
hours, minutes = divmod(minutes, 60)
days, hours = divmod(hours, 24)
days, hours, minutes = list(map(int, (days, hours, minutes)))
seconds = round(seconds, 6)
# Build date
date_str = ""
if days:
date_str = "%sD" % days
# Build time
time_str = "T"
# Hours
bigger_exists = date_str or hours
if bigger_exists:
time_str += "{:02}H".format(hours)
# Minutes
bigger_exists = bigger_exists or minutes
if bigger_exists:
time_str += "{:02}M".format(minutes)
# Seconds
try:
if seconds.is_integer():
seconds_string = "{:02}".format(int(seconds))
else:
# 9 chars long w/ leading 0, 6 digits after decimal
seconds_string = "%09.6f" % seconds
# Remove trailing zeros
seconds_string = seconds_string.rstrip("0")
except AttributeError: # int.is_integer() raises
seconds_string = "{:02}".format(seconds)
time_str += "{}S".format(seconds_string)
return "P" + date_str + time_str
def _datetime_as_isostr(dt: typing.Union[datetime.datetime, datetime.date, datetime.time, datetime.timedelta]) -> str:
'''Converts a datetime.(datetime|date|time|timedelta) object into an ISO 8601 formatted string.
:param dt: The datetime object to convert
:type dt: datetime.datetime or datetime.date or datetime.time or datetime.timedelta
:return: An ISO 8601 formatted string representing the datetime object
:rtype: str
'''
# First try datetime.datetime
if hasattr(dt, "year") and hasattr(dt, "hour"):
dt = typing.cast(datetime.datetime, dt)
# astimezone() fails for naive times in Python 2.7, so make make sure dt is aware (tzinfo is set)
if not dt.tzinfo:
iso_formatted = dt.replace(tzinfo=datetime.timezone.utc).isoformat()
else:
iso_formatted = dt.astimezone(datetime.timezone.utc).isoformat()
# Replace the trailing "+00:00" UTC offset with "Z" (RFC 3339: https://www.ietf.org/rfc/rfc3339.txt)
return iso_formatted.replace("+00:00", "Z")
# Next try datetime.date or datetime.time
try:
dt = typing.cast(typing.Union[datetime.date, datetime.time], dt)
return dt.isoformat()
# Last, try datetime.timedelta
except AttributeError:
dt = typing.cast(datetime.timedelta, dt)
return _timedelta_as_isostr(dt)
class _AspireJSONEncoder(json.JSONEncoder):
'''A JSON encoder that's capable of serializing datetime objects and bytes.'''
def default(self, o: typing.Any) -> typing.Any:
'''Override the default method to handle datetime and bytes serialization.
:param o: The object to serialize.
:type o: Any
:return: A JSON-serializable representation of the object.
:rtype: Any
'''
if isinstance(o, ReferenceExpression):
return o.to_json()
if isinstance(o, _ReferenceHandle):
return o.handle.to_json()
if isinstance(o, Handle):
return o.to_json()
if isinstance(o, (bytes, bytearray)):
return base64.b64encode(o).decode()
try:
return _datetime_as_isostr(o)
except AttributeError:
pass
return super().default(o)
class AspireClient:
'''Client for connecting to the Aspire AppHost via socket/named pipe (synchronous with threads).'''
# Default heartbeat interval in seconds
DEFAULT_HEARTBEAT_INTERVAL = 5.0
def __init__(self, socket_path: str, *, debug: bool | None = None, heartbeat_interval: float | None = None) -> None:
self.socket_path = socket_path
self.debug = debug if debug is not None else False
self._socket: _PipeSocket | None = None
self._request_id = 0
self._pending_requests: dict[int, threading.Event] = {}
self._pending_results: dict[int, tuple[typing.Any, Exception | None]] = {}
self._disconnect_callbacks: list[typing.Callable[[], None]] = []
self._receive_thread: threading.Thread | None = None
self._heartbeat_thread: threading.Thread | None = None
self._cancellation_threads: dict[str, tuple[typing.Callable[[], None], threading.Thread]] = {}
self._cancellation_id = 0
self._heartbeat_interval = heartbeat_interval if heartbeat_interval is not None else self.DEFAULT_HEARTBEAT_INTERVAL
self._heartbeat_stop_event = threading.Event()
self._connected = False
self._connection_error: ConnectionError | None = None
self._lock = threading.Lock()
self._callback_registry: dict[str, typing.Callable[..., typing.Any]] = {}
self._callback_id_counter = 0
self._callback_threads: dict[str, threading.Thread] = {}
def on_disconnect(self, callback: typing.Callable[[], None]) -> None:
'''Register a callback to be called when the connection is lost'''
with self._lock:
self._disconnect_callbacks.append(callback)
def _handle_sigint(self, signum: int, frame: typing.Any) -> None:
'''Handle SIGINT (Ctrl+C) by cancelling all pending requests and closing connection.'''
# Close the connection
self._close_connection(ConnectionError("Shutdown by user"))
# TODO: This will just shutdown the process, which may be best as this will likely
# be run with aspire.run, which will prefer a quick exit with zero exit code.
# Not entirely sure if this signal will be propagated correctly by .NET.... well see.
sys.exit(0)
def _close_connection(self, error: ConnectionError | None = None) -> None:
'''
Central method for closing the connection. Thread-safe and idempotent.
Args:
error: If provided, this is an unexpected disconnection and callbacks
will be notified. If None, this is an intentional disconnect.
This method:
- Closes the socket
- Sets _connected = False
- Stores the error (if provided and not already set)
- Signals the heartbeat thread to stop
- Notifies disconnect callbacks (only on first unexpected disconnection)
'''
should_notify = False
callbacks: list[typing.Callable[[], None]] = []
with self._lock:
# Close socket if open
if self._socket is not None:
try:
self._socket.close()
except Exception:
pass
self._socket = None
# Mark as disconnected
should_notify = self._connected
self._connected = False
callbacks = list(self._disconnect_callbacks)
if should_notify:
if error:
_logger.error("Closing connection to AppHost with error: %s", error)
else:
_logger.info("Closing connection to AppHost")
# Handle error state
if error is not None:
# Unexpected disconnection - store error if first failure
if self._connection_error is None:
self._connection_error = error
else:
# Intentional disconnect - clear any previous error?
self._connection_error = None
for event in self._pending_requests.values():
event.set()
for (cancellation, _ ) in self._cancellation_threads.values():
# Threads will exit on their own when they notice disconnection
cancellation()
# Signal heartbeat to stop (outside lock, it's thread-safe)
self._heartbeat_stop_event.set()
# Notify callbacks outside lock to avoid deadlocks
if should_notify:
for callback in callbacks:
try:
callback()
except Exception:
pass
def connect(self, timeout_ms: int = 10000) -> None:
'''Connect to the Aspire AppHost'''
timeout_sec = timeout_ms / 1000.0
socket = _connect_pipe(self.socket_path, timeout_sec)
with self._lock:
self._socket = socket
self._connected = True
self._connection_error = None
self._heartbeat_stop_event.clear()
# Install SIGINT handler for clean Ctrl+C handling
signal.signal(signal.SIGINT, self._handle_sigint)
_logger.info("Connected to AppHost")
# Start receiving messages in a background thread
self._receive_thread = threading.Thread(target=self._receive_loop, daemon=True)
self._receive_thread.start()
# Start heartbeat thread to monitor connection health
self._heartbeat_thread = threading.Thread(target=self._heartbeat_loop, daemon=True)
self._heartbeat_thread.start()
def _recv_exactly(self, n: int) -> bytes:
'''Read exactly n bytes from the socket'''
data = b""
while len(data) < n:
chunk = typing.cast(_PipeSocket, self._socket).recv(n - len(data))
if not chunk:
raise ConnectionError("Connection closed")
data += chunk
return data
def _read_line(self) -> bytes:
'''Read a line ending with \\r\\n from the socket'''
line = b""
while True:
byte = typing.cast(_PipeSocket, self._socket).recv(1)
if not byte:
raise ConnectionError("Connection closed")
line += byte
if line.endswith(b"\r\n"):
return line[:-2] # Remove \r\n
def _read_headers(self) -> dict[str, str]:
'''Read HTTP-style headers until empty line.
Enforces limits on header count and total size to prevent
memory exhaustion from a malicious peer.
'''
headers: dict[str, str] = {}
total_bytes = 0
while True:
line = self._read_line()
if not line: # Empty line signals end of headers
break
total_bytes += len(line) + 2 # account for \r\n
if len(headers) >= _MAX_HEADER_COUNT:
raise ConnectionError(f"Too many headers (limit {_MAX_HEADER_COUNT})")
if total_bytes > _MAX_HEADER_BYTES:
raise ConnectionError(f"Headers too large (limit {_MAX_HEADER_BYTES} bytes)")
# Parse "Header-Name: value"
if b":" in line:
name, value = line.split(b":", 1)
headers[name.decode("utf-8").strip().lower()] = value.decode("utf-8").strip()
return headers
def _receive_loop(self) -> None:
'''Receive and process messages from the server'''
try:
while self._connected and self._socket:
# Read HTTP-style headers (HeaderDelimitedMessageHandler format)
headers = self._read_headers()
content_length = int(headers.get("content-length", "0"))
if content_length == 0:
continue
if content_length > _MAX_MESSAGE_SIZE:
raise ConnectionError(
f"Message too large: {content_length} bytes "
f"(limit {_MAX_MESSAGE_SIZE} bytes)"
)
# Read message content
message_bytes = self._recv_exactly(content_length)
message_str = message_bytes.decode("utf-8")
message = json.loads(message_str)
if self.debug:
if message.get("result") == "pong":
_logger.debug("<- %s", message)
else:
_logger.info("<- %s", message)
# Handle response or request
if "method" in message:
# This is a request from the server (callback invocation)
self._handle_server_request(message)
elif "id" in message:
# This is a response to our request
request_id = message["id"]
with self._lock:
if request_id in self._pending_requests:
event = self._pending_requests[request_id]
if "error" in message:
self._pending_results[request_id] = (None, Exception(message["error"]["message"]))
else:
self._pending_results[request_id] = (message.get("result"), None)
event.set()
except AttributeError:
# This probably means the socket was closed
pass
except ConnectionError as e:
self._close_connection(e)
except Exception as e:
self._close_connection(ConnectionError(f"Receive loop error: {e}"))
def _heartbeat_loop(self) -> None:
'''Periodically ping the server to check connection health.'''
while not self._heartbeat_stop_event.wait(timeout=self._heartbeat_interval):
with self._lock:
if not self._connected:
break
try:
self.ping()
except Exception as e:
self._close_connection(ConnectionError(f"Heartbeat failed: {e}"))
break
def _handle_server_request(self, message: dict[str, typing.Any]) -> None:
'''Handle a request from the server (e.g., callback invocation)'''
method = message.get("method")
request_id = message.get("id")
params = message.get("params", [])
if method == "invokeCallback":
_logger.debug("Invoking callback with params: %s", params)
callback_id = str(params[0]) if len(params) > 0 else None
args = params[1] if len(params) > 1 else None
# Spawn a separate thread to handle the callback so receive_loop isn't blocked
thread_id = f"cb_thread_{request_id}_{callback_id}"
thread = threading.Thread(
target=self._execute_callback_thread,
args=(callback_id, args, request_id, thread_id),
daemon=True
)
with self._lock:
self._callback_threads[thread_id] = thread
thread.start()
def _execute_callback_thread(
self,
callback_id: str | None,
args: typing.Any,
request_id: int | None,
thread_id: str
) -> None:
'''Execute a callback in a separate thread and send the response.'''
result = None
error = None
try:
try:
with self._lock:
callback = self._callback_registry.get(callback_id) if callback_id else None
if callback:
result = callback(args, self)
_logger.debug("Callback result: %s", result)
else:
error = {"code": -32601, "message": f"Callback not found: {callback_id}"}
except OperationCancelled as e:
_logger.debug("Callback cancelled: %s", e)
return
except Exception as e:
_logger.warning("Exception in callback: %s", e)
# Include type and origin (file:line) for diagnosability,
# but omit the full stack trace and error message to avoid
# leaking sensitive internal details to the server.
tb = e.__traceback__
if tb is not None:
# Walk to the innermost frame (actual error site)
while tb.tb_next is not None:
tb = tb.tb_next
filename = tb.tb_frame.f_code.co_filename
# Send only the basename to avoid leaking full filesystem paths
basename = filename.rsplit("/", 1)[-1].rsplit("\\", 1)[-1]
lineno = tb.tb_lineno
location = f" at {basename}:{lineno}"
else:
location = ""
error = {"code": -32603, "message": f"Internal callback error: {type(e).__name__}{location}"}
# Send response
if request_id is not None:
if error:
response = {
"jsonrpc": "2.0",
"id": request_id,
"error": error
}
else:
response = {
"jsonrpc": "2.0",
"id": request_id,
"result": result
}
try:
self._send_message(response)
except Exception as e:
_logger.error("Failed to send callback response: %s", e)
finally:
# Clean up thread reference
with self._lock:
self._callback_threads.pop(thread_id, None)
def _send_message(self, message: dict[str, typing.Any]) -> None:
'''Send a JSON-RPC message to the server using header-delimited format'''
if self.debug:
if message.get("method") == "ping":
_logger.debug("-> %s", message)
else:
_logger.info("-> %s", message)
message_str = json.dumps(message, cls=_AspireJSONEncoder)
message_bytes = message_str.encode("utf-8")
content_length = len(message_bytes)
# Send with HTTP-style headers (HeaderDelimitedMessageHandler format)
header = f"Content-Length: {content_length}\r\n\r\n"
header_bytes = header.encode("utf-8")
with self._lock:
typing.cast(_PipeSocket, self._socket).sendall(header_bytes + message_bytes)
def _check_connection(self) -> None:
'''Check if connected and raise stored connection error if present.'''
with self._lock:
if self._connection_error:
raise self._connection_error
if not self._connected:
raise RuntimeError("Not connected to AppHost")
def ping(self) -> str:
'''Ping the server'''
self._check_connection()
return self._send_request("ping")
def authenticate(self, token: str) -> None:
'''Authenticate to the AppHost server with a session token.'''
if not bool(self._send_request("authenticate", token)):
raise RuntimeError("Failed to authenticate to the AppHost server.")
def invoke_capability(
self,
capability_id: str,
args: dict[str, typing.Any] | None = None,
kwargs: typing.Mapping[str, typing.Any] | None = None
) -> typing.Any:
'''
Invoke an ATS capability by ID.
Capabilities are operations exposed by [AspireExport] attributes.
Results are automatically wrapped in Handle objects when applicable.
'''
self._check_connection()
result = self._send_request("invokeCapability", capability_id, self._marshal_transport_value(args or {}))
# Check for structured error response
if _is_ats_error(result):
if result["$error"].get("code") == AtsErrorCodes.TYPE_MISMATCH:
raise _format_type_error(result["$error"])
raise AspireError(result["$error"])
# Wrap handles automatically
return _wrap_if_handle(result, self, kwargs)
def _marshal_transport_value(self, value: typing.Any) -> typing.Any:
if callable(value):
return self.register_callback(value)
if isinstance(value, dict):
return {key: self._marshal_transport_value(nested_value) for key, nested_value in value.items()}
if isinstance(value, (list, tuple)):
return [self._marshal_transport_value(item) for item in value]
return value
def _send_request(self, method: str, *params: typing.Any) -> typing.Any:
'''Send a JSON-RPC request and wait for response'''
with self._lock:
self._request_id += 1
request_id = self._request_id
request = {
"jsonrpc": "2.0",
"id": request_id,
"method": method,
"params": list(params) if params else []
}
# Create event for response
event = threading.Event()
with self._lock:
self._pending_requests[request_id] = event
# Send request
self._send_message(request)
# Wait for response
event.wait()
# Get result
with self._lock:
if request_id not in self._pending_results:
# Request was cancelled/interrupted
if self._connection_error:
raise self._connection_error
raise RuntimeError("Request was cancelled")
del self._pending_requests[request_id]
result, error = self._pending_results.pop(request_id)
if error:
raise error
return result
def register_cancellation_token(self, cancellation_timeout: int | None) -> str | None:
if not cancellation_timeout:
return None
with self._lock:
cancellation_id = f"ct_{self._cancellation_id}_{int(time.time() * 1000)}"
cancellation_token = threading.Event()
def cancellation_thread():
cancellation_token.wait()
self._check_connection()
# Send cancellation request to server
try:
self._send_request("cancelToken", cancellation_id)
except Exception:
pass # Ignore errors during cancellation
self._cancellation_threads.pop(cancellation_id, None)
thread = threading.Thread(target=cancellation_thread, daemon=True)
cancel_timer = threading.Timer(cancellation_timeout, cancellation_token.set)
def shutdown():
cancel_timer.cancel()
cancellation_token.set()
self._cancellation_threads[cancellation_id] = (shutdown, thread)
self._cancellation_id += 1
thread.start()
cancel_timer.start()
return cancellation_id
def disconnect(self) -> None:
'''Disconnect from the server'''
self._close_connection(error=None) # Intentional disconnect, no error
if self._heartbeat_thread and self._heartbeat_thread.is_alive():
self._heartbeat_thread.join(timeout=1.0)
if self._receive_thread and self._receive_thread.is_alive():
self._receive_thread.join(timeout=1.0)
# Wait for any pending callback threads to finish
with self._lock:
callback_threads = list(self._callback_threads.values())
for thread in callback_threads:
if thread.is_alive():
thread.join(timeout=1.0)
for (_, thread) in self._cancellation_threads.values():
if thread.is_alive():
thread.join(timeout=1.0)
def register_callback(self, callback: typing.Callable[..., typing.Any] | None) -> str | None:
'''
Register a callback function that can be invoked from the .NET side.
Returns a callback ID that should be passed to methods accepting callbacks.
.NET passes arguments as an object with positional keys: { p0: value0, p1: value1, ... }
This function automatically extracts positional parameters and wraps handles.
'''
if callback is None:
return None
with self._lock:
self._callback_id_counter += 1
callback_id = f"callback_{secrets.token_hex(16)}"
def wrapper(args: typing.Any, client: AspireClient) -> typing.Any:
# .NET sends args as object { p0: value0, p1: value1, ... }
if isinstance(args, dict):
arg_array = []
i = 0
while True:
key = f"p{i}"
if key in args:
arg_array.append(_wrap_if_handle(args[key], client))
i += 1
else:
break
if arg_array:
return callback(*arg_array)
# No args or null
if args is None:
return callback()
# Single primitive value (shouldn't happen with current protocol)
return callback(_wrap_if_handle(args, client))
with self._lock:
self._callback_registry[callback_id] = wrapper
return callback_id
@property
def connected(self) -> bool:
'''Check if connected to the server'''
return self._connected
# ============================================================================
# CancellationToken
# ============================================================================
class CancellationToken:
'''Represents a cancellation token that can be used to cancel a callback in progress.'''
handle: Handle
def __init__(self, handle: Handle, client: AspireClient) -> None:
self.handle = handle
self._client = client
def cancel(self) -> None:
'''Cancel the token, which will signal the server to cancel the associated operation.'''
self._client._send_request("cancelToken", self.handle.handle_id)
raise OperationCancelled(self.handle.handle_id)
# ============================================================================
# Reference Expression
# ============================================================================
class ReferenceExpression:
'''Represents a reference expression passed to capabilities.
Supports both value mode (format + valueProviders) and conditional mode (condition + whenTrue + whenFalse).
'''
def __init__(self, handle: Handle | None, **kwargs) -> None:
'''
Creates a reference expression from a format string and value providers.
Args:
format_str: Format string with {0}, {1}, etc. placeholders
*value_providers: Handles to value providers
Returns:
A ReferenceExpression instance
'''
self._handle = handle
self._format = kwargs.get("format_str")
self._value_providers = kwargs.get("value_providers", [])
self._condition = kwargs.get("condition")
self._when_true = kwargs.get("when_true")
self._when_false = kwargs.get("when_false")
self._match_value = kwargs.get("match")
@classmethod
def format_string(cls, format_str: str, *value_providers: typing.Any) -> "ReferenceExpression":
'''
Creates a reference expression from a format string and value providers.
Args:
format_str: Format string with {0}, {1}, etc. placeholders
*value_providers: Handles to value providers
Returns:
A ReferenceExpression instance
'''
providers = [_extract_handle_for_expr(v) for v in value_providers]
return cls(None, format_str=format_str, value_providers=providers)
@classmethod
def conditional(cls, condition: typing.Any, **kwargs) -> "ReferenceExpression":
'''
Creates a conditional reference expression.
Args:
condition: The condition to evaluate
match: The value to match against
when_true: ReferenceExpression if condition matches
when_false: ReferenceExpression if condition does not match
Returns:
A ReferenceExpression instance
'''
return cls(None, condition=condition, **kwargs)
def to_json(self) -> typing.Mapping[str, typing.Any]:
'''
Serializes the reference expression for JSON-RPC transport.
Uses the $expr format recognized by the server.
'''
if self._handle:
return self._handle.to_json()
if self._condition and self._when_true and self._when_false and self._match_value is not None:
return {
"$expr": {
"condition": self._condition,
"whenTrue": self._when_true.to_json(),
"whenFalse": self._when_false.to_json(),
"matchValue": self._match_value,
}
}
if self._format:
result: dict[str, typing.Any] = {
"$expr": {
"format": self._format,
}
}
if self._value_providers:
result["$expr"]["valueProviders"] = self._value_providers
return result
raise ValueError("Invalid ReferenceExpression: must have either handle, condition, or format")
def __repr__(self) -> str:
if self._handle:
return f"ReferenceExpression(handle={self._handle.handle_id})"
if self._condition:
return "ReferenceExpression(conditional)"
return f"ReferenceExpression(formattedString)"
def _extract_handle_for_expr(value: typing.Any) -> typing.Any:
'''
Extracts a value for use in reference expressions.
Supports handles (objects) and string literals.
'''
if value is None:
raise ValueError("Cannot use None in reference expression")
# String literals - include directly in the expression
if isinstance(value, str):
return value
# Number literals - convert to string
if isinstance(value, (int, float)):
return str(value)
# Handle objects - get their JSON representation
if isinstance(value, (Handle, _ReferenceHandle)):
return value
# Objects with $handle property (already in handle format)
if isinstance(value, dict) and "$handle" in value:
return value
raise ValueError(
f"Cannot use value of type {type(value).__name__} in reference expression. "
f"Expected a handle object, string, or number."
)
def string_expr(value: str, **kwargs: typing.Any) -> ReferenceExpression:
'''
Helper function for creating reference expressions with named placeholders.
Use this to create dynamic expressions that reference endpoints, parameters, and other
value providers. The expression is evaluated at runtime by Aspire.
Example:
```python
redis = await builder.add_redis("cache")
endpoint = await redis.get_endpoint("tcp")
# Create a reference expression using named placeholders
expr = string_expr("redis://{host}:{port}", host=endpoint, port=6379)
# Use it in an environment variable
await api.with_environment("REDIS_URL", expr)
```
'''
# Replace named placeholders with numbered ones
value_providers = []
format_str = value
for key, value in kwargs.items():
placeholder = f"{{{key}}}"
if placeholder in format_str:
index = len(value_providers)
format_str = format_str.replace(placeholder, f"{{{index}}}")
value_providers.append(value)
return ReferenceExpression.format_string(format_str, *value_providers)
def conditional_expr(condition: typing.Any, *, match: str, when_true: str | ReferenceExpression, when_false: str | ReferenceExpression) -> ReferenceExpression:
'''
Helper function for creating conditional reference expressions.
This allows you to create dynamic expressions that evaluate conditions at runtime.
Example:
```python
# Create a conditional expression based on an environment variable
condition = {"$handle": "Aspire.Hosting.EnvironmentVariableReference:ENVIRONMENT"}
expr = conditional_expr(condition, match="production",
when_true="redis://prod-redis:6379",
when_false="redis://dev-redis:6379"
)
await api.with_environment("REDIS_URL", expr)
```
'''
if isinstance(when_true, str):
when_true = ReferenceExpression.format_string(when_true)
if isinstance(when_false, str):
when_false = ReferenceExpression.format_string(when_false)
return ReferenceExpression.conditional(condition, match=match, when_true=when_true, when_false=when_false)
# ============================================================================
# AspireList[T] - Mutable List Wrapper
# ============================================================================
TItem = typing.TypeVar("TItem")
class AspireList(typing.MutableSequence[TItem]):
'''
Wrapper for a mutable .NET List<T>.
Example:
```python
items = await resource.get_items() # Returns AspireList[ItemBuilder]
# Modify the list
items.append(new_item)
items[0] = another_item
del items[1]
```
'''
def __init__(
self,
handle: Handle,
client: AspireClient,
) -> None:
self._handle = handle
self._client = client
# ---- Required abstract methods from MutableSequence ----
def __len__(self) -> int:
'''Gets the number of elements in the list.'''
result = self._client.invoke_capability(
"Aspire.Hosting/List.length",
{"list": self._handle}
)
return int(result)
@typing.overload
def __getitem__(self, index: int) -> TItem: ...
@typing.overload
def __getitem__(self, index: slice) -> list[TItem]: ...
def __getitem__(self, index: int | slice) -> TItem | list[TItem]:
'''Gets the element at the specified index or a slice of elements.'''
if not isinstance(index, int):
raise NotImplementedError("Get by slice not yet supported.")
result = self._client.invoke_capability(
"Aspire.Hosting/List.get",
{"list": self._handle, "index": index}
)
return list(result)
@typing.overload
def __setitem__(self, index: int, value: TItem) -> None: ...
@typing.overload
def __setitem__(self, index: slice, value: typing.Iterable[TItem]) -> None: ...
def __setitem__(self, index: int | slice, value: TItem | typing.Iterable[TItem]) -> None:
'''Sets the element at the specified index or replaces a slice of elements.'''
if not isinstance(index, int):
raise NotImplementedError("Set by slice not yet supported.")
self._client.invoke_capability(
"Aspire.Hosting/List.set",
{"list": self._handle, "index": index, "value": value}
)
@typing.overload
def __delitem__(self, index: int) -> None: ...
@typing.overload
def __delitem__(self, index: slice) -> None: ...
def __delitem__(self, index: int | slice) -> None:
'''Deletes the element at the specified index or a slice of elements.'''
if not isinstance(index, int):
raise NotImplementedError("Delete by slice not yet supported.")
self._client.invoke_capability(
"Aspire.Hosting/List.removeAt",
{"list": self._handle, "index": index}
)
def insert(self, index: int, value: TItem) -> None:
'''Inserts an element at the specified index.'''
self._client.invoke_capability(
"Aspire.Hosting/List.set",
{"list": self._handle, "index": index, "item": value}
)
def __repr__(self) -> str:
'''Returns a string representation of the list.'''
return f"AspireList(handle={self._handle.handle_id})"
# ============================================================================
# AspireDict[K, V] - Mutable Dictionary Wrapper
# ============================================================================
TKey = typing.TypeVar("TKey")
TValue = typing.TypeVar("TValue")
class AspireDict(typing.MutableMapping[TKey, TValue]):
'''
Wrapper for a mutable .NET Dictionary<K, V>.
Example:
```python
config = await resource.get_config() # Returns AspireDict[str, str]
# Modify the dict
config["key"] = "value"
del config["old_key"]
```
'''
def __init__(
self,
handle: Handle,
client: AspireClient,
) -> None:
self._handle = handle
self._client = client
def __len__(self) -> int:
'''Gets the number of key-value pairs in the dictionary.'''
result = self._client.invoke_capability(
"Aspire.Hosting/Dict.count",
{"dict": self._handle}
)
return int(result)
def __getitem__(self, key: TKey) -> TValue:
'''Gets the value associated with the specified key.'''
result = self._client.invoke_capability(
"Aspire.Hosting/Dict.get",
{"dict": self._handle, "key": key}
)
if result is None:
raise KeyError(key)
return result
def __setitem__(self, key: TKey, value: TValue) -> None:
'''Sets the value for the specified key.'''
self._client.invoke_capability(
"Aspire.Hosting/Dict.set",
{"dict": self._handle, "key": key, "value": value}
)
def __delitem__(self, key: TKey) -> None:
'''Removes the key-value pair with the specified key.'''
result = self._client.invoke_capability(
"Aspire.Hosting/Dict.remove",
{"dict": self._handle, "key": key}
)
if result is False:
raise KeyError(key)
def __iter__(self) -> typing.Iterator[TKey]:
'''Returns an iterator over the keys.'''
result = self._client.invoke_capability(
"Aspire.Hosting/Dict.keys",
{"dict": self._handle}
)
return iter(result)
def __contains__(self, key: object) -> bool:
result = self._client.invoke_capability(
"Aspire.Hosting/Dict.has",
{"dict": self._handle, "key": key}
)
return result
def __repr__(self) -> str:
'''Returns a string representation of the dictionary.'''
return f"AspireDict(handle={self._handle.handle_id})"
def clear(self) -> None:
self._client.invoke_capability(
"Aspire.Hosting/Dict.clear",
{"dict": self._handle}
)
def _validate_type(arg: typing.Any, expected_type: typing.Any) -> bool:
if typing.get_origin(expected_type) is typing.Iterable:
if isinstance(arg, str):
return False
item_type = typing.get_args(expected_type)[0]
if not isinstance(arg, typing.Iterable):
return False
for item in arg:
if not _validate_type(item, item_type):
return False
elif typing.get_origin(expected_type) is typing.Mapping:
key_type, value_type = typing.get_args(expected_type)
if not isinstance(arg, typing.Mapping):
return False
for key, value in arg.items():
if not _validate_type(key, key_type):
return False
if not _validate_type(value, value_type):
return False
elif typing.get_origin(expected_type) is typing.Callable:
return callable(arg)
elif isinstance(arg, (tuple, typing.Mapping)):
return False
elif typing.get_origin(expected_type) is typing.Literal:
if arg not in typing.get_args(expected_type):
return False
elif expected_type is None:
if arg is not None:
return False
elif subtypes := typing.get_args(expected_type):
# This is probably a Union type
return any([_validate_type(arg, subtype) for subtype in subtypes])
elif not isinstance(arg, expected_type):
return False
return True
def _validate_tuple_types(args: typing.Any, arg_types: tuple[typing.Any, ...]) -> bool:
if not isinstance(args, tuple):
return False
if len(args) != len(arg_types):
return False
for arg, expected_type in zip(args, arg_types):
if not _validate_type(arg, expected_type):
return False
return True
def _validate_dict_types(args: typing.Any, arg_types: typing.Any) -> bool:
if not isinstance(args, typing.Mapping):
return False
type_hints = typing.get_type_hints(arg_types, include_extras=True)
for key, expected_type in type_hints.items():
if typing.get_origin(expected_type) is typing.Required:
expected_type = typing.get_args(expected_type)[0]
if key not in args:
return False
if key not in args:
continue
value = args[key]
if not _validate_type(value, expected_type):
return False
return True
""";
/// <summary>
/// The <c>DistributedApplicationBuilder</c> Python class definition for the generated SDK.
/// </summary>
public const string DistributedApplicationBuilder = """
class DistributedApplicationBuilder:
'''Type class for DistributedApplicationBuilder.'''
def __init__(self, client: AspireClient, options: CreateBuilderOptions) -> None:
self._handle = None
self._client = client
self._options = options
@property
def handle(self) -> Handle:
'''Gets the underlying handle for the builder.'''
if not self._handle:
raise RuntimeError("Builder connection not initialized.")
return self._handle
def __enter__(self) -> DistributedApplicationBuilder:
self._handle = self._client.invoke_capability(
'Aspire.Hosting/createBuilder',
{'argsOrOptions': self._options}
)
return self
def __exit__(self, exc_type, exc_value, traceback) -> None:
self._client.disconnect()
def run(self, *, timeout: int | None = None) -> None:
'''Builds and runs the distributed application.'''
app = self.build()
app.run(timeout=timeout)
""";
/// <summary>
/// Connection helper code for creating the Aspire client and builder.
/// </summary>
public const string ConnectionHelperCode = """
def _get_client(*, debug: bool, heartbeat_interval: int | None) -> AspireClient:
'''
Creates and connects to the Aspire AppHost.
Reads connection info from environment variables set by `aspire run`.
'''
socket_path = os.environ.get('REMOTE_APP_HOST_SOCKET_PATH')
if not socket_path:
raise ValueError(
'REMOTE_APP_HOST_SOCKET_PATH environment variable not set. '
'Run this application using `aspire run`.'
)
client = AspireClient(socket_path, debug=debug, heartbeat_interval=heartbeat_interval)
client.connect()
auth_token = os.environ.get('ASPIRE_REMOTE_APPHOST_TOKEN')
if not auth_token:
raise ValueError(
'ASPIRE_REMOTE_APPHOST_TOKEN environment variable not set. '
'Run this application using `aspire run`.'
)
client.authenticate(auth_token)
return client
def create_builder(
*,
args: typing.Iterable[str] | None = None,
project_directory: str | None = None,
app_host_file_path: str | None = None,
container_registry_override: str | None = None,
disable_dashboard: bool | None = None,
dashboard_application_name: str | None = None,
allow_unsecured_transport: bool | None = None,
enable_resource_logging: bool | None = None,
options: CreateBuilderOptions | None = None,
debug: bool | None = None,
heartbeat_interval: int | None = None,
) -> AbstractContextManager[DistributedApplicationBuilder]:
'''
Creates a new distributed application builder.
This is the entry point for building Aspire applications.
Args:
args (Iterable[str]): Command-line arguments to pass to the AppHost. By default, this will be set to any additional arguments
passed to the Aspire command line (arguments specified after '--'). Specifying them here will override that default.
project_directory (str): The directory containing the AppHost project file. By default, this will use the ASPIRE_PROJECT_DIRECTORY
environment variable if set, otherwise it will use the current working directory.
app_host_file_path (str): The path to the AppHost source file. By default, this will use the ASPIRE_APPHOST_FILEPATH
environment variable if set.
container_registry_override (str): When containers are used, use this value to override the container registry.
disable_dashboard (bool): Determines whether the dashboard is disabled.
dashboard_application_name (str): The application name to display in the dashboard.
allow_unsecured_transport (bool): Allows the use of HTTP urls for the AppHost resource endpoint.
enable_resource_logging (bool): Enables resource logging.
options (CreateBuilderOptions): An optional dict containing any of the above options. Specifying options here will override default behaviours,
but individual parameters will take precedence.
debug (bool): Whether to enable logging of the communication between the client and AppHost server.
Default behaviour will be determined by whether `--debug` is passed as an Aspire command-line argument, or
if the ASPIRE_DEBUG environment variable is set. Enabling or disabling here will override those defaults.
Messages will be logged as INFO, with the 'aspire_app' logger name (connection heartbeat messages will be logged at DEBUG).
heartbeat_interval (int): Optional interval in seconds for sending heartbeat messages to the AppHost. Default value is 5 seconds.
Returns:
A DistributedApplicationBuilder instance
'''
is_debug = debug if debug is not None else os.environ.get('ASPIRE_DEBUG', 'false').lower() == 'true'
client = _get_client(debug=is_debug, heartbeat_interval=heartbeat_interval)
# Default args and project_directory if not provided
effective_options = options or CreateBuilderOptions()
if args is not None:
effective_options['Args'] = args
elif not effective_options.get('Args'):
effective_options['Args'] = sys.argv[1:]
if project_directory is not None:
effective_options['ProjectDirectory'] = project_directory
elif not effective_options.get('ProjectDirectory'):
effective_options['ProjectDirectory'] = os.environ.get('ASPIRE_PROJECT_DIRECTORY', os.getcwd())
if app_host_file_path is not None:
effective_options['AppHostFilePath'] = app_host_file_path
elif not effective_options.get('AppHostFilePath'):
app_host_file_path = os.environ.get('ASPIRE_APPHOST_FILEPATH')
if app_host_file_path:
effective_options['AppHostFilePath'] = app_host_file_path
if container_registry_override is not None:
effective_options['ContainerRegistryOverride'] = container_registry_override
if disable_dashboard is not None:
effective_options['DisableDashboard'] = disable_dashboard
if dashboard_application_name is not None:
effective_options['DashboardApplicationName'] = dashboard_application_name
if allow_unsecured_transport is not None:
effective_options['AllowUnsecuredTransport'] = allow_unsecured_transport
if enable_resource_logging is not None:
effective_options['EnableResourceLogging'] = enable_resource_logging
return DistributedApplicationBuilder(client, effective_options)
""";
}