Application Helpers

The SDK is designed to be particularly useful for developing Cumulocity microservices. For this, the module provides two helper classes that take care of microservice specific authentication.

The SimpleCumulocityApp class should be used for single tenant microservices. It automatically reads the microservice's environment to determines the microservice access credentials.

The MultiTenantCumulocityApp class should be used for multi-tenant microservices which need to handle requests for arbitrary Cumulocity tenants. It reads the microservice's environment to determine the necessary bootstrap credentials and provides additional functions to dynamically obtain CumulocityClient instances for specific tenants.

For interactive use (e.g. in Jupyter Notebooks) the get_client function can be used to obtain a ready to use CumulocityClient instance. It honors from standard Cumulocity environment variables for not explicity provided connection details, and falls back to interactive prompts for anything still missing.

---

SimpleCumulocityApp

Application-like Cumulocity API.

The SimpleCumulocityApp class is intended to be used as base within a single-tenant microservice hosted on Cumulocity. It evaluates the environment to the resolve the authentication information automatically.

Note: This class should be used in Cumulocity microservices using the PER_TENANT authentication mode only. It will not function in environments using the MULTITENANT mode.

The SimpleCumulocityApp class is an enhanced version of the standard CumulocityClient class. All Cumulocity functions can be used directly. Additionally, it can be used to provide CumulocityClient instances for specific named users via the get_user_instance function.

__init__(application_key: str | None = None, processing_mode: str | None = None, cache_size: int = 100, cache_ttl: int = 3600, max_concurrent: int | None = None)

Create a new tenant specific instance.

Parameters:

Name	Type	Description	Default
application_key	str	An application key to include in all requests for tracking purposes; this will be read from the environment (APPLICATION_KEY) if not defined.	None
cache_size	int	The maximum number of cached user instances (if user instances are created at all).	100
cache_ttl	int	An maximum cache time for user instances (if user instances are created at all).	3600
max_concurrent	int	The maximum number of concurrent tasks handed down to the underlying tenant.	None

Returns:

Type	Description
	A new CumulocityApp instance

close() async

Close the app's own session and all cached per-user sessions.

MultiTenantCumulocityApp

Multi-tenant enabled Cumulocity application wrapper.

The MultiTenantCumulocityApp class is intended to be used as base within a multi-tenant microservice hosted on Cumulocity. It evaluates the environment to the resolve the bootstrap authentication information automatically.

Note: This class is intended to be used in Cumulocity microservices using the MULTITENANT authentication mode. It will not function in PER_TENANT environments.

The MultiTenantCumulocityApp class serves as a factory. It provides access to tenant-specific CumulocityApi instances via the get_tenant_instance function. A special bootstrap CumulocityApi instance is available via the bootstrap_instance property.

__init__(application_key: str | None = None, processing_mode: str | None = None, cache_size: int = 100, cache_ttl: int = 3600, connection_limit: int = 100, connection_limit_per_host: int = 0, max_concurrent: int | None = None)

Create a new instance.

Parameters:

Name	Type	Description	Default
application_key	str | None	An application key to include in all requests for tracking purposes; this will be read from the environment (APPLICATION_KEY) if not defined.	None
cache_size	int | None	The maximum number of cached tenant instances (if tenant instances are created at all).	100
cache_ttl	int | None	An maximum cache time for tenant instances (if tenant instances are created at all).	3600
connection_limit	int	Maximum number of simultaneous HTTP connections in the shared pool across all tenant/user clients.	100
connection_limit_per_host	int	Maximum simultaneous connections to a single host; 0 means unlimited (bounded only by connection_limit).	0
max_concurrent	int	The maximum number of concurrent tasks across all subscribed tenants.	None

Returns:

Type	Description
	A new MultiTenantCumulocityApp instance

close() async

Close all sessions and the shared connection pool.

create_listener(callback: Callable[[set[str]], None] | None = None, sequential: bool = False, polling_interval: float = 3600, startup_delay: float = 60) -> SubscriptionListener

Create a subscription listener for this app.

Parameters:

Name	Type	Description	Default
callback	Callable	an async callback function to be invoked when the subscribers change; Use the add_callback function to add callbacks for individually added/removed subscribers.	None
sequential	bool	If True, callbacks run one at a time (guarded by an asyncio.Lock). Default False — callbacks run concurrently as asyncio tasks.	False
polling_interval	float	How often to poll for changes.	3600
startup_delay	float	How many seconds to wait after a subscriber change before invoking the callbacks.	60

Returns:

Type	Description
SubscriptionListener	A SubscriptionListener instance.

get_subscribers() -> list[str] async

Query the subscribed tenants.

Returns:

Type	Description
list[str]	A list of tenant ID.

get_tenant_instance(tenant_id: str | None = None, headers: Mapping[str, str] | None = None, cookies: Mapping[str, str] | None = None) -> CumulocityClient async

Provide access to a tenant-specific instance in a multi-tenant application setup.

Parameters:

Name	Type	Description	Default
tenant_id	str	ID of the tenant to get access to	None
headers	Mapping	Inbound request headers, the tenant ID is resolved from the Authorization header	None
cookies	Mapping	A dictionary of HTTP Cookie entries. The user access is based on an authorization cookie as provided by Cumulocity.	None

Returns:

Type	Description
CumulocityClient	A CumulocityApi instance authorized for a tenant user

get_client(base_url: str | None = None, tenant_id: str | None = None, username: str | None = None, password: str | None = None, max_concurrent: int | None = None, log_level: int | str | None = None) -> CumulocityClient async

Get a ready to use CumulocityClient instance for use in interactive sessions.

Reads connection details from standard¹ Cumulocity environment variables when not explicitly provided, and falls back to interactive prompts for anything still missing.

c8y = await get_client()

Parameters:

Name	Type	Description	Default
base_url	str	Cumulocity base URL; reads C8Y_BASEURL if omitted.	None
tenant_id	str	Tenant ID; reads C8Y_TENANT if omitted.	None
username	str	Username; reads C8Y_USER if omitted.	None
password	str	Password; reads C8Y_PASSWORD if omitted.	None
max_concurrent	int	Maximum number of concurrent tasks.	None
log_level	int | str	If set, installs a colorized stdout handler on the root logger. Looser-than-default levels (DEBUG, INFO) are applied to the pyc8y logger only, so third-party libs stay quiet; stricter-than-default levels (ERROR, CRITICAL) are applied to the root logger to dampen everything. Enable other namespaces on demand with e.g. logging.getLogger("aiohttp").setLevel(logging.DEBUG).	None

¹ See also the go-c8y-cli (https://goc8ycli.netlify.app/docs/concepts/sessions/#continuous-integration-usage-environment-variables) and Cumulocity microservice bootstrap (https://cumulocity.com/docs/microservice-sdk/general-aspects/#microservice-bootstrap)