Skip to content

ContainerAppClient API Reference

cfa.cloudops.ContainerAppClient

Client for managing Azure Container Apps jobs via the Azure SDK.

Provides methods to list, start, and inspect jobs in a resource group using managed identity authentication. Supports job info retrieval, command inspection, job existence checks, and flexible job start options.

Source code in cfa/cloudops/_containerappclient.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 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
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
class ContainerAppClient:
    """
    Client for managing Azure Container Apps jobs via the Azure SDK.

    Provides methods to list, start, and inspect jobs in a resource group using
    managed identity authentication. Supports job info retrieval, command inspection,
    job existence checks, and flexible job start options.
    """

    def __init__(
        self,
        dotenv_path=None,
        resource_group=None,
        subscription_id=None,
        job_name=None,
    ):
        """
        Initialize a ContainerAppClient for Azure Container Apps jobs.

        Args:
            dotenv_path (str, optional): Path to a .env file to load environment variables.
            resource_group (str, optional): Azure resource group name. If None, uses env var AZURE_RESOURCE_GROUP_NAME.
            subscription_id (str, optional): Azure subscription ID. If None, uses env var AZURE_SUBSCRIPTION_ID.
            job_name (str, optional): Job name for Container App Job.

        Raises:
            ValueError: If required parameters are missing and not set in environment variables.
        """
        self.credential = ManagedIdentityCredential()
        dotenv.load_dotenv(dotenv_path)
        sub_c = SubscriptionClient(self.credential)
        # pull in account info and save to environment vars
        account_info = list(sub_c.subscriptions.list())[0]
        os.environ["AZURE_SUBSCRIPTION_ID"] = account_info.subscription_id
        os.environ["AZURE_TENANT_ID"] = account_info.tenant_id
        os.environ["AZURE_RESOURCE_GROUP_NAME"] = account_info.display_name
        if resource_group is None:
            resource_group = os.getenv("AZURE_RESOURCE_GROUP_NAME")
            if resource_group is None:
                raise ValueError(
                    "No resource_group provided and no RESOURCE_GROUP env var found."
                )
        self.resource_group = resource_group
        if subscription_id is None:
            subscription_id = os.getenv("AZURE_SUBSCRIPTION_ID")
            if subscription_id is None:
                raise ValueError(
                    "No subscription_id provided and no AZURE_SUBSCRIPTION_ID env var found."
                )
        self.subscription_id = subscription_id
        self.job_name = job_name

        self.client = ContainerAppsAPIClient(
            credential=self.credential, subscription_id=subscription_id
        )
        logger.debug("client initialized.")

    def get_job_info(self, job_name: str | None = None):
        """
        Retrieve detailed information about a specific Container App job.

        Args:
            job_name (str): Name of the job to retrieve information for.

        Returns:
            dict: Dictionary containing job details.
        """
        if job_name is None:
            if self.job_name is None:
                raise ValueError("Please specify a job name.")
            else:
                job_name = self.job_name

        for i in self.client.jobs.list_by_resource_group(self.resource_group):
            if i.name == job_name:
                job_info = i
        return job_info.as_dict()

    def get_command_info(self, job_name: str | None = None):
        """
        Get command, image, and environment details for containers in a job.

        Args:
            job_name (str): Name of the job to inspect.

        Returns:
            list[dict]: List of container info dicts (name, image, command, args, env).
        """
        if job_name is None:
            if self.job_name is None:
                raise ValueError("Please specify a job name.")
            else:
                job_name = self.job_name

        for i in self.client.jobs.list_by_resource_group(self.resource_group):
            if i.name == job_name:
                job_info = i
        c_info = job_info.__dict__["template"].__dict__["containers"]
        container_dicts = []
        for c in c_info:
            container_dict = {
                "job_name": c.name,
                "image": c.image,
                "command": c.command,
                "args": c.args,
                "env": c.env,
            }
            container_dicts.append(container_dict)
        return container_dicts

    def list_jobs(self):
        """
        List all Container App job names in the resource group.

        Returns:
            list[str]: List of job names.
        """
        job_list = [
            i.name
            for i in self.client.jobs.list_by_resource_group(
                self.resource_group
            )
        ]
        return job_list

    def check_job_exists(self, job_name: str):
        """
        Check if a Container App job exists in the resource group.

        Args:
            job_name (str): Name of the job to check.

        Returns:
            bool: True if job exists, False otherwise.
        """
        if job_name in self.list_jobs():
            return True
        else:
            logger.info(f"Container App Job {job_name} not found.")
            return False

    def start_job(
        self,
        job_name: str | None = None,
        command: list[str] | None = None,
        args: list[str] | None = None,
        env: list[str] | None = None,
    ):
        """
        Start a Container App job, optionally overriding command, args, or environment.

        Args:
            job_name (str, optional): Name of the job to start. If None, uses default job_name.
            command (list[str], optional): Command to run in the container.
            args (list[str], optional): Arguments for the command.
            env (list[str], optional): Environment variables for the container.

        Raises:
            ValueError: If required parameters are missing or not in correct format.
        """
        if job_name is None:
            if self.job_name is None:
                raise ValueError("Please specify a job name.")
            else:
                job_name = self.job_name
        if not command and not args and not env:
            logger.debug("submitting job start request.")
            self.client.jobs.begin_start(
                resource_group_name=self.resource_group, job_name=job_name
            )
        else:
            # raise error if command/args/env not lists
            if command is not None and not isinstance(command, list):
                raise ValueError("Command must be in list format.")
            if args is not None and not isinstance(args, list):
                raise ValueError("Args must be in list format.")
            if env is not None and not isinstance(env, list):
                raise ValueError("Env must be in list format.")
            new_containers = []
            for i in self.client.jobs.list_by_resource_group(
                self.resource_group
            ):
                if i.name == job_name:
                    job_info = i
            for c in job_info.__dict__["template"].__dict__["containers"]:
                image = c.image
                name = c.name
                resources = c.resources
                container = JobExecutionContainer(
                    image=image,
                    name=name,
                    command=command,
                    args=args,
                    env=env,
                    resources=resources,
                )
                new_containers.append(container)
            t = JobExecutionTemplate(containers=new_containers)
            logger.debug("submitting job start request.")
            try:
                self.client.jobs.begin_start(
                    resource_group_name=self.resource_group,
                    job_name=job_name,
                    template=t,
                )
                print(f"Started job {job_name}.")
            except Exception as e:
                logger.error(f"Failed to start job {job_name}: {e}")
                raise

    def stop_job(self, job_name: str, job_execution_name: str):
        """
        Stop a specific execution of an Azure Container App Job.

        Args:
            job_name (str): Name of the Container App Job.
            job_execution_name (str): Name of the job execution to stop.

        Returns:
            Any: Response object from the Azure SDK if successful, or None if an error occurs.

        Raises:
            Exception: If the stop operation fails.
        """
        try:
            response = self.client.jobs.begin_stop_execution(
                resource_group_name=self.resource_group,
                job_name=job_name,
                job_execution_name=job_execution_name,
            ).result()
            logger.info(
                f"Job execution '{job_execution_name}' for job '{job_name}' stopped successfully."
            )
            return response
        except Exception as e:
            logger.error(f"Error stopping job execution: {e}")
            return None

__init__(dotenv_path=None, resource_group=None, subscription_id=None, job_name=None)

Initialize a ContainerAppClient for Azure Container Apps jobs.

Parameters:

Name Type Description Default
dotenv_path str

Path to a .env file to load environment variables.

 None
resource_group str

Azure resource group name. If None, uses env var AZURE_RESOURCE_GROUP_NAME.

 None
subscription_id str

Azure subscription ID. If None, uses env var AZURE_SUBSCRIPTION_ID.

 None
job_name str

Job name for Container App Job.

 None

Raises:

Type Description
ValueError

If required parameters are missing and not set in environment variables.
Source code in cfa/cloudops/_containerappclient.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
63
64
65
66
67
68
69
70
71
def __init__(
    self,
    dotenv_path=None,
    resource_group=None,
    subscription_id=None,
    job_name=None,
):
    """
    Initialize a ContainerAppClient for Azure Container Apps jobs.

    Args:
        dotenv_path (str, optional): Path to a .env file to load environment variables.
        resource_group (str, optional): Azure resource group name. If None, uses env var AZURE_RESOURCE_GROUP_NAME.
        subscription_id (str, optional): Azure subscription ID. If None, uses env var AZURE_SUBSCRIPTION_ID.
        job_name (str, optional): Job name for Container App Job.

    Raises:
        ValueError: If required parameters are missing and not set in environment variables.
    """
    self.credential = ManagedIdentityCredential()
    dotenv.load_dotenv(dotenv_path)
    sub_c = SubscriptionClient(self.credential)
    # pull in account info and save to environment vars
    account_info = list(sub_c.subscriptions.list())[0]
    os.environ["AZURE_SUBSCRIPTION_ID"] = account_info.subscription_id
    os.environ["AZURE_TENANT_ID"] = account_info.tenant_id
    os.environ["AZURE_RESOURCE_GROUP_NAME"] = account_info.display_name
    if resource_group is None:
        resource_group = os.getenv("AZURE_RESOURCE_GROUP_NAME")
        if resource_group is None:
            raise ValueError(
                "No resource_group provided and no RESOURCE_GROUP env var found."
            )
    self.resource_group = resource_group
    if subscription_id is None:
        subscription_id = os.getenv("AZURE_SUBSCRIPTION_ID")
        if subscription_id is None:
            raise ValueError(
                "No subscription_id provided and no AZURE_SUBSCRIPTION_ID env var found."
            )
    self.subscription_id = subscription_id
    self.job_name = job_name

    self.client = ContainerAppsAPIClient(
        credential=self.credential, subscription_id=subscription_id
    )
    logger.debug("client initialized.")

check_job_exists(job_name)

Check if a Container App job exists in the resource group.

Parameters:

Name Type Description Default
job_name str

Name of the job to check.

 required

Returns:

Name Type Description
bool

True if job exists, False otherwise.
Source code in cfa/cloudops/_containerappclient.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def check_job_exists(self, job_name: str):
    """
    Check if a Container App job exists in the resource group.

    Args:
        job_name (str): Name of the job to check.

    Returns:
        bool: True if job exists, False otherwise.
    """
    if job_name in self.list_jobs():
        return True
    else:
        logger.info(f"Container App Job {job_name} not found.")
        return False

get_command_info(job_name=None)

Get command, image, and environment details for containers in a job.

Parameters:

Name Type Description Default
job_name str

Name of the job to inspect.

 None

Returns:

Type Description

list[dict]: List of container info dicts (name, image, command, args, env).
Source code in cfa/cloudops/_containerappclient.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def get_command_info(self, job_name: str | None = None):
    """
    Get command, image, and environment details for containers in a job.

    Args:
        job_name (str): Name of the job to inspect.

    Returns:
        list[dict]: List of container info dicts (name, image, command, args, env).
    """
    if job_name is None:
        if self.job_name is None:
            raise ValueError("Please specify a job name.")
        else:
            job_name = self.job_name

    for i in self.client.jobs.list_by_resource_group(self.resource_group):
        if i.name == job_name:
            job_info = i
    c_info = job_info.__dict__["template"].__dict__["containers"]
    container_dicts = []
    for c in c_info:
        container_dict = {
            "job_name": c.name,
            "image": c.image,
            "command": c.command,
            "args": c.args,
            "env": c.env,
        }
        container_dicts.append(container_dict)
    return container_dicts

get_job_info(job_name=None)

Retrieve detailed information about a specific Container App job.

Parameters:

Name Type Description Default
job_name str

Name of the job to retrieve information for.

 None

Returns:

Name Type Description
dict

Dictionary containing job details.
Source code in cfa/cloudops/_containerappclient.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def get_job_info(self, job_name: str | None = None):
    """
    Retrieve detailed information about a specific Container App job.

    Args:
        job_name (str): Name of the job to retrieve information for.

    Returns:
        dict: Dictionary containing job details.
    """
    if job_name is None:
        if self.job_name is None:
            raise ValueError("Please specify a job name.")
        else:
            job_name = self.job_name

    for i in self.client.jobs.list_by_resource_group(self.resource_group):
        if i.name == job_name:
            job_info = i
    return job_info.as_dict()

list_jobs()

List all Container App job names in the resource group.

Returns:

Type Description

list[str]: List of job names.
Source code in cfa/cloudops/_containerappclient.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def list_jobs(self):
    """
    List all Container App job names in the resource group.

    Returns:
        list[str]: List of job names.
    """
    job_list = [
        i.name
        for i in self.client.jobs.list_by_resource_group(
            self.resource_group
        )
    ]
    return job_list

start_job(job_name=None, command=None, args=None, env=None)

Start a Container App job, optionally overriding command, args, or environment.

Parameters:

Name Type Description Default
job_name str

Name of the job to start. If None, uses default job_name.

 None
command list[str]

Command to run in the container.

 None
args list[str]

Arguments for the command.

 None
env list[str]

Environment variables for the container.

 None

Raises:

Type Description
ValueError

If required parameters are missing or not in correct format.
Source code in cfa/cloudops/_containerappclient.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def start_job(
    self,
    job_name: str | None = None,
    command: list[str] | None = None,
    args: list[str] | None = None,
    env: list[str] | None = None,
):
    """
    Start a Container App job, optionally overriding command, args, or environment.

    Args:
        job_name (str, optional): Name of the job to start. If None, uses default job_name.
        command (list[str], optional): Command to run in the container.
        args (list[str], optional): Arguments for the command.
        env (list[str], optional): Environment variables for the container.

    Raises:
        ValueError: If required parameters are missing or not in correct format.
    """
    if job_name is None:
        if self.job_name is None:
            raise ValueError("Please specify a job name.")
        else:
            job_name = self.job_name
    if not command and not args and not env:
        logger.debug("submitting job start request.")
        self.client.jobs.begin_start(
            resource_group_name=self.resource_group, job_name=job_name
        )
    else:
        # raise error if command/args/env not lists
        if command is not None and not isinstance(command, list):
            raise ValueError("Command must be in list format.")
        if args is not None and not isinstance(args, list):
            raise ValueError("Args must be in list format.")
        if env is not None and not isinstance(env, list):
            raise ValueError("Env must be in list format.")
        new_containers = []
        for i in self.client.jobs.list_by_resource_group(
            self.resource_group
        ):
            if i.name == job_name:
                job_info = i
        for c in job_info.__dict__["template"].__dict__["containers"]:
            image = c.image
            name = c.name
            resources = c.resources
            container = JobExecutionContainer(
                image=image,
                name=name,
                command=command,
                args=args,
                env=env,
                resources=resources,
            )
            new_containers.append(container)
        t = JobExecutionTemplate(containers=new_containers)
        logger.debug("submitting job start request.")
        try:
            self.client.jobs.begin_start(
                resource_group_name=self.resource_group,
                job_name=job_name,
                template=t,
            )
            print(f"Started job {job_name}.")
        except Exception as e:
            logger.error(f"Failed to start job {job_name}: {e}")
            raise

stop_job(job_name, job_execution_name)

Stop a specific execution of an Azure Container App Job.

Parameters:

Name Type Description Default
job_name str

Name of the Container App Job.

 required
job_execution_name str

Name of the job execution to stop.

 required

Returns:

Name Type Description
Any

Response object from the Azure SDK if successful, or None if an error occurs.

Raises:

Type Description
Exception

If the stop operation fails.
Source code in cfa/cloudops/_containerappclient.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
def stop_job(self, job_name: str, job_execution_name: str):
    """
    Stop a specific execution of an Azure Container App Job.

    Args:
        job_name (str): Name of the Container App Job.
        job_execution_name (str): Name of the job execution to stop.

    Returns:
        Any: Response object from the Azure SDK if successful, or None if an error occurs.

    Raises:
        Exception: If the stop operation fails.
    """
    try:
        response = self.client.jobs.begin_stop_execution(
            resource_group_name=self.resource_group,
            job_name=job_name,
            job_execution_name=job_execution_name,
        ).result()
        logger.info(
            f"Job execution '{job_execution_name}' for job '{job_name}' stopped successfully."
        )
        return response
    except Exception as e:
        logger.error(f"Error stopping job execution: {e}")
        return None