Skip to content

Reference

load_config(config_file, verbose=True, sandbox='prod')

Initialize the api from a config file

Examples:

>>> import aezpz
>>> api = aezpz.load_config("auth.json")

Parameters:

Name Type Description Default
config_file str

The filepath of your json config file that you downloaded from AEP

 required
verbose bool

Whether to print the status code of every request. Defaults to True

 True
sandbox str

The name of the sandbox to use. Defaults to 'prod'

 'prod'

Returns:

Type Description
Api

The initialized api interface
Source code in aezpz/api.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def load_config(config_file: str, verbose: bool=True, sandbox: str='prod') -> Api:
    """ Initialize the api from a config file

    Examples:
        >>> import aezpz
        >>> api = aezpz.load_config("auth.json")

    Args:
        config_file: The filepath of your json config file that you downloaded from AEP
        verbose: Whether to print the status code of every request. Defaults to True
        sandbox: The name of the sandbox to use. Defaults to 'prod'

    Returns:
        The initialized api interface
    """
    return Api(config_file, verbose=verbose, sandbox=sandbox)

Api

The main interface to the Adobe XDM API

Attributes:

Name Type Description
base_url str

The base url of the Adobe XDM API. Defaults to 'https://platform.adobe.io'
headers dict

The default headers to be sent with every request.
verbose bool

Whether to print the status code of every request. Defaults to True
sandbox str

The name of the sandbox to use. Defaults to 'prod'
registry ResourceCollection

A collection of all resources in all containers
global_registry ResourceCollection

A collection of all resources in the global container
tenant_registry ResourceCollection

A collection of all resources in the tenant container
schemas SchemaCollection

A collection of all schemas in the tenant container
global_schemas SchemaCollection

A collection of all schemas in the global container
tenant_schemas SchemaCollection

A collection of all schemas in the tenant container
classes ClassCollection

A collection of all classes in the tenant container
global_classes ClassCollection

A collection of all classes in the global container
tenant_classes ClassCollection

A collection of all classes in the tenant container
field_groups FieldGroupCollection

A collection of all field groups in the tenant container
global_field_groups FieldGroupCollection

A collection of all field groups in the global container
tenant_field_groups FieldGroupCollection

A collection of all field groups in the tenant container
data_types DataTypeCollection

A collection of all data types in the tenant container
global_data_types DataTypeCollection

A collection of all data types in the global container
tenant_data_types DataTypeCollection

A collection of all data types in the tenant container
behaviors BehaviorCollection

A collection of all behaviors in the tenant container

Methods:

Name Description
ref

Retrieves the value associated with the given reference.
request

The underlying method for all requests to the api.

Examples:

Set the sandbox to 'stage'

>>> api.sandbox = 'stage'
>>> api.headers
{ 'x-sandbox-name': 'stage' }
Source code in aezpz/api.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
 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
class Api:
    """The main interface to the Adobe XDM API

    Attributes:
        base_url: The base url of the Adobe XDM API. Defaults to 'https://platform.adobe.io'
        headers: The default headers to be sent with every request.
        verbose: Whether to print the status code of every request. Defaults to True
        sandbox: The name of the sandbox to use. Defaults to 'prod'

        registry: A collection of all resources in all containers
        global_registry: A collection of all resources in the global container
        tenant_registry: A collection of all resources in the tenant container
        schemas: A collection of all schemas in the tenant container
        global_schemas: A collection of all schemas in the global container
        tenant_schemas: A collection of all schemas in the tenant container
        classes: A collection of all classes in the tenant container
        global_classes: A collection of all classes in the global container
        tenant_classes: A collection of all classes in the tenant container
        field_groups: A collection of all field groups in the tenant container
        global_field_groups: A collection of all field groups in the global container
        tenant_field_groups: A collection of all field groups in the tenant container
        data_types: A collection of all data types in the tenant container
        global_data_types: A collection of all data types in the global container
        tenant_data_types: A collection of all data types in the tenant container
        behaviors: A collection of all behaviors in the tenant container

    Methods:
        ref: Retrieves the value associated with the given reference.
        request: The underlying method for all requests to the api.

    Examples:
        Set the sandbox to 'stage'
        >>> api.sandbox = 'stage'
        >>> api.headers
        { 'x-sandbox-name': 'stage' }
    """

    base_url: str
    sandbox: str
    verbose: bool
    _access_token: str
    _config: dict

    registry: schema.ResourceCollection
    global_registry: schema.ResourceCollection
    tenant_registry: schema.ResourceCollection
    schemas: schema.SchemaCollection
    global_schemas: schema.SchemaCollection
    tenant_schemas: schema.SchemaCollection
    classes: schema.ClassCollection
    global_classes: schema.ClassCollection
    tenant_classes: schema.ClassCollection
    field_groups: schema.FieldGroupCollection
    global_field_groups: schema.FieldGroupCollection
    tenant_field_groups: schema.FieldGroupCollection
    data_types: schema.DataTypeCollection
    global_data_types: schema.DataTypeCollection
    tenant_data_types: schema.DataTypeCollection
    behaviors: schema.BehaviorCollection
    datasets: datasets.DatasetCollection
    batches: datasets.BatchCollection

    def __init__(self, config_file, verbose=True, sandbox='prod'):
        self.sandbox = sandbox
        self.verbose = verbose
        self.base_url = 'https://platform.adobe.io'
        self._config = self.load_config_file(config_file)
        self._access_token = self.authenticate()
        self.registry = schema.ResourceCollection(self)
        self.global_registry = schema.ResourceCollection(self, container='global')
        self.tenant_registry = schema.ResourceCollection(self, container='tenant')
        self.schemas = schema.SchemaCollection(self)
        self.global_schemas = schema.SchemaCollection(self, container='global')
        self.tenant_schemas = schema.SchemaCollection(self, container='tenant')
        self.classes = schema.ClassCollection(self)
        self.global_classes = schema.ClassCollection(self, container='global')
        self.tenant_classes = schema.ClassCollection(self, container='tenant')
        self.field_groups = schema.FieldGroupCollection(self)
        self.global_field_groups = schema.FieldGroupCollection(self, container='global')
        self.tenant_field_groups = schema.FieldGroupCollection(self, container='tenant')
        self.data_types = schema.DataTypeCollection(self)
        self.global_data_types = schema.DataTypeCollection(self, container='global')
        self.tenant_data_types = schema.DataTypeCollection(self, container='tenant')
        self.behaviors = schema.BehaviorCollection(self)

        self.datasets = datasets.DatasetCollection(self)
        self.batches = datasets.BatchCollection(self)

    def ref(self, ref: str) -> schema.Resource:
        """
        Retrieves the value associated with the given reference.

        Args:
            ref: The `$id` or `meta:altId` of the reference to retrieve.

        Returns:
            The value associated with the reference.

        Examples:
            >>> api.ref('https://ns.adobe.com/xdm/context/profile')
            <Class xdm.context.profile>

            >>> api.ref('_mytenant.schemas.7a5416d135713dae7957')
            <Schema 7a5416d135713dae7957>
        """
        return self.registry.get(ref)

    @property
    def headers(self) -> dict:
        assert getattr(self, 'config', None) and getattr(self, 'access_token', None), 'need to authenticate first'
        return {
            'x-sandbox-name': self.sandbox,
            'x-api-key': self._config['CLIENT_ID'],
            'x-gw-ims-org-id': self._config['ORG_ID'],
            'Authorization': 'Bearer ' + self._access_token,
        }

    @property
    def me(self) -> str:
        return self._config['ACCOUNT_ID']

    def load_config_file(self, config_file) -> dict:
        config = json.load(Path(config_file).open('r'))
        return {
            'CLIENT_ID': config['CLIENT_ID'],
            'ORG_ID': config['ORG_ID'],
            'CLIENT_SECRET': config['CLIENT_SECRETS'][0],
            'SCOPES': config['SCOPES'],
            'ORG_ID': config['ORG_ID'],
            'ACCOUNT_ID': config['TECHNICAL_ACCOUNT_ID'],
        }

    def authenticate(self) -> str:
        r = requests.post('https://ims-na1.adobelogin.com/ims/token/v2', params={
            'grant_type': 'client_credentials',
            'client_id': self._config['CLIENT_ID'],
            'client_secret': self._config['CLIENT_SECRET'],
            'scope': ','.join(self._config['SCOPES'])
        })
        r.raise_for_status()
        return r.json()['access_token']

    def request(self, method, path, headers={}, **kwargs) -> Optional[dict]:
        """
        The underlying method for all requests to the api. Wraps the requests library.

        Args:
            method (str): The HTTP method to use for the request (e.g., 'GET', 'POST', 'PUT', 'DELETE').
            path (str): The path of the resource to request.
            headers (dict, optional): Additional headers to include in the request. Defaults to an empty dictionary.
            **kwargs: Additional keyword arguments to pass to the underlying requests library.

        Returns:
            dict: The JSON response from the server, if any.

        Raises:
            requests.exceptions.HTTPError: If the response status code indicates an error.

        Examples:
            >>> api.request('GET', '/data/foundation/schemaregistry/global/behaviors', headers={'Accept': 'application/vnd.adobe.xed-id+json'})
            { "results": [{ "$id": "https://ns.adobe.com/xdm/data/time-series" }, ...], "_page": { "count": 3 } }
        """
        assert 'Authorization' in self.headers, 'need to load_config first'
        r = requests.request(
            method=method,
            url=self.base_url+path,
            headers={
                **self.headers,
                **headers,
            },
            **kwargs,
        )
        if self.verbose:
            print(r.status_code, r.request.method, r.request.path_url)
        if not r.ok:
            try:
                error = r.json()
                if 'title' in error:
                    print(error['title'])
                if 'detail' in error:
                    print(error['detail'])
            except:
                print(r.text)
        r.raise_for_status()
        if len(r.content):
            return r.json()

ref(ref)

Retrieves the value associated with the given reference.

Parameters:

Name Type Description Default
ref str

The $id or meta:altId of the reference to retrieve.

 required

Returns:

Type Description
Resource

The value associated with the reference.

Examples:

>>> api.ref('https://ns.adobe.com/xdm/context/profile')
<Class xdm.context.profile>

>>> api.ref('_mytenant.schemas.7a5416d135713dae7957')
<Schema 7a5416d135713dae7957>
Source code in aezpz/api.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def ref(self, ref: str) -> schema.Resource:
    """
    Retrieves the value associated with the given reference.

    Args:
        ref: The `$id` or `meta:altId` of the reference to retrieve.

    Returns:
        The value associated with the reference.

    Examples:
        >>> api.ref('https://ns.adobe.com/xdm/context/profile')
        <Class xdm.context.profile>

        >>> api.ref('_mytenant.schemas.7a5416d135713dae7957')
        <Schema 7a5416d135713dae7957>
    """
    return self.registry.get(ref)

request(method, path, headers={}, **kwargs)

The underlying method for all requests to the api. Wraps the requests library.

Parameters:

Name Type Description Default
method str

The HTTP method to use for the request (e.g., 'GET', 'POST', 'PUT', 'DELETE').

 required
path str

The path of the resource to request.

 required
headers dict

Additional headers to include in the request. Defaults to an empty dictionary.

 {}
**kwargs

Additional keyword arguments to pass to the underlying requests library.

 {}

Returns:

Name Type Description
dict Optional[dict]

The JSON response from the server, if any.

Raises:

Type Description
HTTPError

If the response status code indicates an error.

Examples:

>>> api.request('GET', '/data/foundation/schemaregistry/global/behaviors', headers={'Accept': 'application/vnd.adobe.xed-id+json'})
{ "results": [{ "$id": "https://ns.adobe.com/xdm/data/time-series" }, ...], "_page": { "count": 3 } }
Source code in aezpz/api.py 
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
def request(self, method, path, headers={}, **kwargs) -> Optional[dict]:
    """
    The underlying method for all requests to the api. Wraps the requests library.

    Args:
        method (str): The HTTP method to use for the request (e.g., 'GET', 'POST', 'PUT', 'DELETE').
        path (str): The path of the resource to request.
        headers (dict, optional): Additional headers to include in the request. Defaults to an empty dictionary.
        **kwargs: Additional keyword arguments to pass to the underlying requests library.

    Returns:
        dict: The JSON response from the server, if any.

    Raises:
        requests.exceptions.HTTPError: If the response status code indicates an error.

    Examples:
        >>> api.request('GET', '/data/foundation/schemaregistry/global/behaviors', headers={'Accept': 'application/vnd.adobe.xed-id+json'})
        { "results": [{ "$id": "https://ns.adobe.com/xdm/data/time-series" }, ...], "_page": { "count": 3 } }
    """
    assert 'Authorization' in self.headers, 'need to load_config first'
    r = requests.request(
        method=method,
        url=self.base_url+path,
        headers={
            **self.headers,
            **headers,
        },
        **kwargs,
    )
    if self.verbose:
        print(r.status_code, r.request.method, r.request.path_url)
    if not r.ok:
        try:
            error = r.json()
            if 'title' in error:
                print(error['title'])
            if 'detail' in error:
                print(error['detail'])
        except:
            print(r.text)
    r.raise_for_status()
    if len(r.content):
        return r.json()

ResourceCollection

Base class for all resource collections. Can be used directly through the registry attribute of the API Instance to retrieve resources of any type

Methods:

Name Description
get

Retrieves a resource based on the provided reference.
find

Finds a resource based on the specified parameters.
list

Finds all resources based on the specified parameters.

Examples:

>>> api.registry.list()
[<Class xdm.classes.summarymetrics>, <Schema 7a5416d13571>, ...]
Source code in aezpz/schema.py 
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
253
254
255
256
class ResourceCollection:
    """ Base class for all resource collections. 
    Can be used directly through the `registry` attribute of the API Instance to
    retrieve resources of any type

    Methods:
        get: Retrieves a resource based on the provided reference.
        find: Finds a resource based on the specified parameters.
        list: Finds all resources based on the specified parameters.

    Examples:
        >>> api.registry.list()
        [<Class xdm.classes.summarymetrics>, <Schema 7a5416d13571>, ...]
    """
    api: Api
    container: Optional[Container]
    resources: list[ResourceType]

    def __init__(self, api: Api, container:Optional[Container]=None, resources:list[ResourceType]=[]):
        self.api = api
        if len(resources) == 0:
            resources = [
                ResourceType.DATA_TYPE,
                ResourceType.FIELD_GROUP,
                ResourceType.SCHEMA,
                ResourceType.CLASS,
                ResourceType.BEHAVIOR,
            ]
        assert container is None or container in ('global','tenant')
        for resource in resources:
            assert isinstance(resource, ResourceType)
        self.resources = resources
        self.container = container

    @cached_property
    def containers(self) -> list[Container]:
        containers = ['tenant','global']
        if self.container is not None:
            containers = [ self.container ]
        return containers

    def get(self, ref: str) -> Resource:
        """
        Retrieves a resource based on the provided reference.

        Args:
            ref: The `$id` or `meta:altId` of the reference to retrieve.

        Returns:
            Resource: The retrieved resource.
        """
        ref = SchemaRef(ref)
        assert ref.resource in self.resources
        return ref.init(self.api)

    def find(self,
             full: bool = True,
             **params) -> Resource:
        """
        Finds a resource based on the specified parameters.

        Args:
            full: If True will use `vnd.adobe.xed-full+json` accept header.
            **params: Additional parameters for filtering the resources.

        Returns:
            Resource: The found resource.

        Raises:
            Exception: If no resource is found or multiple resources match the parameters.

        Examples:
            >>> api.registry.find(title='My Schema')
            <Schema 7a5416d13571 title="My Schema" version="1.0">
        """
        resources = self.list(full=full, **params)
        if len(resources) == 0:
            raise Exception(f'Could not find resource')
        if len(resources) > 1:
            raise Exception(f'Multiple resources match the parameters')
        return resources[0]

    def _paginate(self, container, resource, full: bool = False, query: dict = {}) -> list[dict]:
        records = []
        params = {}
        if len(query):
            params['property'] = ','.join(
                k + '==' + v
                for k,v in query.items()
            )
        more = True
        while more:
            r = self.api.request('GET',
                                 path=get_resource_path(container, resource),
                                 headers=get_accept_header(
                                    xed='full' if full else None,
                                    xed_version=None
                                 ),
                                 params=params)
            records += r['results']
            if r['_page'].get('next') is not None:
                params['start'] = r['_page']['next']
            else:
                more = False
        return records

    def list(self,
               full: bool = False,
               **query
            ) -> list[Resource]:
        """
        Finds all resources based on the specified parameters.

        Args:
            full: If True will use `vnd.adobe.xed-full+json` accept header. Defaults to True.
            **query: Additional query parameters for filtering the resources.

        Returns:
            List[Resource]: The list of found resources.

        Examples:
            >>> api.registry.list()
            [<Class xdm.classes.summarymetrics>, <Schema 7a5416d13571>, ...]
        """
        results = []
        for resource in self.resources:
            for container in self.containers:
                for record in self._paginate(container, resource, full, query=query):
                    results.append(resource._class(self.api, record))
        return results

    def _create(self, body) -> Resource:
        if self.container == 'global':
            raise Exception('cannot create global resource')
        assert len(self.resources) == 1
        r = self.api.request('POST',
            path=get_resource_path(
                container='tenant',
                resource=self.resources[0],
            ),
            headers=get_accept_header(),
            json=body
        )
        return self.resources[0]._class(self.api, r)

get(ref)

Retrieves a resource based on the provided reference.

Parameters:

Name Type Description Default
ref str

The $id or meta:altId of the reference to retrieve.

 required

Returns:

Name Type Description
Resource Resource

The retrieved resource.
Source code in aezpz/schema.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
def get(self, ref: str) -> Resource:
    """
    Retrieves a resource based on the provided reference.

    Args:
        ref: The `$id` or `meta:altId` of the reference to retrieve.

    Returns:
        Resource: The retrieved resource.
    """
    ref = SchemaRef(ref)
    assert ref.resource in self.resources
    return ref.init(self.api)

find(full=True, **params)

Finds a resource based on the specified parameters.

Parameters:

Name Type Description Default
full bool

If True will use vnd.adobe.xed-full+json accept header.

 True
**params

Additional parameters for filtering the resources.

 {}

Returns:

Name Type Description
Resource Resource

The found resource.

Raises:

Type Description
Exception

If no resource is found or multiple resources match the parameters.

Examples:

>>> api.registry.find(title='My Schema')
<Schema 7a5416d13571 title="My Schema" version="1.0">
Source code in aezpz/schema.py 
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
def find(self,
         full: bool = True,
         **params) -> Resource:
    """
    Finds a resource based on the specified parameters.

    Args:
        full: If True will use `vnd.adobe.xed-full+json` accept header.
        **params: Additional parameters for filtering the resources.

    Returns:
        Resource: The found resource.

    Raises:
        Exception: If no resource is found or multiple resources match the parameters.

    Examples:
        >>> api.registry.find(title='My Schema')
        <Schema 7a5416d13571 title="My Schema" version="1.0">
    """
    resources = self.list(full=full, **params)
    if len(resources) == 0:
        raise Exception(f'Could not find resource')
    if len(resources) > 1:
        raise Exception(f'Multiple resources match the parameters')
    return resources[0]

list(full=False, **query)

Finds all resources based on the specified parameters.

Parameters:

Name Type Description Default
full bool

If True will use vnd.adobe.xed-full+json accept header. Defaults to True.

 False
**query

Additional query parameters for filtering the resources.

 {}

Returns:

Type Description
list[Resource]

List[Resource]: The list of found resources.

Examples:

>>> api.registry.list()
[<Class xdm.classes.summarymetrics>, <Schema 7a5416d13571>, ...]
Source code in aezpz/schema.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def list(self,
           full: bool = False,
           **query
        ) -> list[Resource]:
    """
    Finds all resources based on the specified parameters.

    Args:
        full: If True will use `vnd.adobe.xed-full+json` accept header. Defaults to True.
        **query: Additional query parameters for filtering the resources.

    Returns:
        List[Resource]: The list of found resources.

    Examples:
        >>> api.registry.list()
        [<Class xdm.classes.summarymetrics>, <Schema 7a5416d13571>, ...]
    """
    results = []
    for resource in self.resources:
        for container in self.containers:
            for record in self._paginate(container, resource, full, query=query):
                results.append(resource._class(self.api, record))
    return results

SchemaCollection

Bases: ResourceCollection

Collection of Schema resources.

Initialized through api.schemas, api.global_schemas, or api.tenant_schemas.

Methods:

Name Description
get

Retrieves a schema based on the provided reference.
find

Finds a schema based on the specified parameters.
list

Finds all schemas based on the specified parameters.
create

Creates a new schema.

Examples:

Get a schema by reference

>>> api.schemas.get('_mytenant.schemas.7a5416d13572')
<Schema 7a5416d13572 title="My Schema" version="1.0">

Find a schema by title

>>> api.tenant_schemas.find(title='My Schema')
<Schema 7a5416d13572 title="My Schema" version="1.0">

List all global and tenant schemas

>>> api.schemas.list()
[<Schema xdm.schemas.computed-attributes>, <Schema 7a5416d13572>, ...]

List all global schemas

>>> api.global_schemas.list()
[<Schema xdm.schemas.computed-attributes>, <Schema xdm.schemas.consentidmap>, ...]

List all tenant schemas

>>> api.tenant_schemas.list()
[<Schema 7a5416d13572>, <Schema 7a5416d13571>, ...]
Source code in aezpz/schema.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
class SchemaCollection(ResourceCollection):
    """ Collection of Schema resources.

    Initialized through `api.schemas`, `api.global_schemas`, or `api.tenant_schemas`.

    Methods:
        get: Retrieves a schema based on the provided reference.
        find: Finds a schema based on the specified parameters.
        list: Finds all schemas based on the specified parameters.
        create: Creates a new schema.

    Examples:
        Get a schema by reference
        >>> api.schemas.get('_mytenant.schemas.7a5416d13572')
        <Schema 7a5416d13572 title="My Schema" version="1.0">

        Find a schema by title
        >>> api.tenant_schemas.find(title='My Schema')
        <Schema 7a5416d13572 title="My Schema" version="1.0">

        List all global and tenant schemas
        >>> api.schemas.list()
        [<Schema xdm.schemas.computed-attributes>, <Schema 7a5416d13572>, ...]

        List all global schemas
        >>> api.global_schemas.list()
        [<Schema xdm.schemas.computed-attributes>, <Schema xdm.schemas.consentidmap>, ...]

        List all tenant schemas
        >>> api.tenant_schemas.list()
        [<Schema 7a5416d13572>, <Schema 7a5416d13571>, ...]
    """
    def __init__(self, api: Api, container:Optional[Container]=None):
        super().__init__(api, container=container, resources=[ResourceType.SCHEMA])

    def get(self, id) -> Schema:
        return super().get(id)

    def find(self, full:bool=True, **params) -> Schema:
        return super().find(full, **params)

    def list(self, full:bool=False, **params) -> list[Schema]:
        return super().list(full, **params)

    def create(
            self,
            title: str,
            parent: Class,
            description: str='',
            field_groups: list[FieldGroup] = [],
        ) -> Schema:
        """
        Create a new schema.

        Args:
            title: The title of the schema.
            parent: The parent class that the schema inherits from.
            description: The description of the schema.
            field_groups: The list of field groups for the schema.

        Returns:
            Schema: The created schema.

        Examples:
            >>> schema = api.schemas.create(
            ...     title='My Schema',
            ...     parent=api.classes.get('_xdm.context.profile'),
            ...     description='My test schema',
            ...     field_groups=[api.field_groups.get('_mytenant.mixins.f7d78220431')]
            ... )
        """
        assert isinstance(parent, Class), 'Must inherit from a class'
        for field_group in field_groups:
            assert isinstance(field_group, FieldGroup)
        return self._create({
            'type': 'object',
            'title': title,
            'description': description,
            'allOf': [
                { '$ref': ref.ref }
                for ref in ([parent] + field_groups)
            ]
        })

create(title, parent, description='', field_groups=[])

Create a new schema.

Parameters:

Name Type Description Default
title str

The title of the schema.

 required
parent Class

The parent class that the schema inherits from.

 required
description str

The description of the schema.

 ''
field_groups list[FieldGroup]

The list of field groups for the schema.

 []

Returns:

Name Type Description
Schema Schema

The created schema.

Examples:

>>> schema = api.schemas.create(
...     title='My Schema',
...     parent=api.classes.get('_xdm.context.profile'),
...     description='My test schema',
...     field_groups=[api.field_groups.get('_mytenant.mixins.f7d78220431')]
... )
Source code in aezpz/schema.py 
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
def create(
        self,
        title: str,
        parent: Class,
        description: str='',
        field_groups: list[FieldGroup] = [],
    ) -> Schema:
    """
    Create a new schema.

    Args:
        title: The title of the schema.
        parent: The parent class that the schema inherits from.
        description: The description of the schema.
        field_groups: The list of field groups for the schema.

    Returns:
        Schema: The created schema.

    Examples:
        >>> schema = api.schemas.create(
        ...     title='My Schema',
        ...     parent=api.classes.get('_xdm.context.profile'),
        ...     description='My test schema',
        ...     field_groups=[api.field_groups.get('_mytenant.mixins.f7d78220431')]
        ... )
    """
    assert isinstance(parent, Class), 'Must inherit from a class'
    for field_group in field_groups:
        assert isinstance(field_group, FieldGroup)
    return self._create({
        'type': 'object',
        'title': title,
        'description': description,
        'allOf': [
            { '$ref': ref.ref }
            for ref in ([parent] + field_groups)
        ]
    })

ClassCollection

Bases: ResourceCollection

Collection of Class resources.

Initialized through api.classes, api.global_classes, or api.tenant_classes.

Methods:

Name Description
get

Retrieves a class based on the provided reference.
find

Finds a class based on the specified parameters.
list

Finds all classes based on the specified parameters.
create

Creates a new class.

Examples:

Get a class by reference

>>> api.classes.get('_mytenant.classes.7a5416d13572')
<Class 7a5416d13572 title="My Class" version="1.0">

Find a class by title

>>> api.tenant_classes.find(title='My Class')
<Class 7a5416d13572 title="My Class" version="1.0">

List all global and tenant classes

>>> api.classes.list()
[<Class xdm.context.profile>, <Class 7a5416d13572>, ...]

List all global classes

>>> api.global_classes.list()
[<Class xdm.context.profile>, <Class xdm.classes.conversion>, ...]

List all tenant classes

>>> api.tenant_classes.list()
[<Class 7a5416d13572>, <Class 7a5416d13571>, ...]
Source code in aezpz/schema.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
class ClassCollection(ResourceCollection):
    """ Collection of Class resources.

    Initialized through `api.classes`, `api.global_classes`, or `api.tenant_classes`.

    Methods:
        get: Retrieves a class based on the provided reference.
        find: Finds a class based on the specified parameters.
        list: Finds all classes based on the specified parameters.
        create: Creates a new class.

    Examples:
        Get a class by reference
        >>> api.classes.get('_mytenant.classes.7a5416d13572')
        <Class 7a5416d13572 title="My Class" version="1.0">

        Find a class by title
        >>> api.tenant_classes.find(title='My Class')
        <Class 7a5416d13572 title="My Class" version="1.0">

        List all global and tenant classes
        >>> api.classes.list()
        [<Class xdm.context.profile>, <Class 7a5416d13572>, ...]

        List all global classes
        >>> api.global_classes.list()
        [<Class xdm.context.profile>, <Class xdm.classes.conversion>, ...]

        List all tenant classes
        >>> api.tenant_classes.list()
        [<Class 7a5416d13572>, <Class 7a5416d13571>, ...]
    """
    def __init__(self, api: Api, container:Optional[Container]=None):
        super().__init__(api, container=container, resources=[ResourceType.CLASS])

    def get(self, id) -> Class:
        return super().get(id)

    def find(self, full:bool=True, **params) -> Class:
        return super().find(full, **params)

    def list(self, full:bool=False, **params) -> list[Class]:
        return super().list(full, **params)

    # TODO: also allow direct definitions of fields
    # TODO: make behavior default to "adhoc" if field_groups have been defined
    def create(
            self,
            title: str,
            behavior: Behavior,
            description: str = '',
            field_groups: list[FieldGroup] = [],
        ) -> Class:
        """
        Create a new class.

        Args:
            title: The title of the class.
            behavior: The behavior of the class.
            description: The description of the class.
            field_groups: The list of field groups for the class.

        Returns:
            Class: The created class.

        Examples:
            >>> my_class = api.classes.create(
            ...     title='My Class',
            ...     behavior=api.behaviors.adhoc,
            ...     description='My test class',
            ...     field_groups=[api.field_groups.get('_mytenant.mixins.f7d78220431')]
            ... )
        """
        assert isinstance(behavior, Behavior), 'Must inherit from a behavior'
        for field_group in field_groups:
            assert isinstance(field_group, FieldGroup)
        return self._create({
            'type': 'object',
            'title': title,
            'description': description,
            'allOf': [
                { '$ref': ref.ref }
                for ref in ([behavior] + field_groups)
            ]
        })

create(title, behavior, description='', field_groups=[])

Create a new class.

Parameters:

Name Type Description Default
title str

The title of the class.

 required
behavior Behavior

The behavior of the class.

 required
description str

The description of the class.

 ''
field_groups list[FieldGroup]

The list of field groups for the class.

 []

Returns:

Name Type Description
Class Class

The created class.

Examples:

>>> my_class = api.classes.create(
...     title='My Class',
...     behavior=api.behaviors.adhoc,
...     description='My test class',
...     field_groups=[api.field_groups.get('_mytenant.mixins.f7d78220431')]
... )
Source code in aezpz/schema.py 
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
def create(
        self,
        title: str,
        behavior: Behavior,
        description: str = '',
        field_groups: list[FieldGroup] = [],
    ) -> Class:
    """
    Create a new class.

    Args:
        title: The title of the class.
        behavior: The behavior of the class.
        description: The description of the class.
        field_groups: The list of field groups for the class.

    Returns:
        Class: The created class.

    Examples:
        >>> my_class = api.classes.create(
        ...     title='My Class',
        ...     behavior=api.behaviors.adhoc,
        ...     description='My test class',
        ...     field_groups=[api.field_groups.get('_mytenant.mixins.f7d78220431')]
        ... )
    """
    assert isinstance(behavior, Behavior), 'Must inherit from a behavior'
    for field_group in field_groups:
        assert isinstance(field_group, FieldGroup)
    return self._create({
        'type': 'object',
        'title': title,
        'description': description,
        'allOf': [
            { '$ref': ref.ref }
            for ref in ([behavior] + field_groups)
        ]
    })

FieldGroupCollection

Bases: ResourceCollection

Collection of FieldGroup resources.

Initialized through api.field_groups, api.global_field_groups, or api.tenant_field_groups.

Methods:

Name Description
get

Retrieves a field group based on the provided reference.
find

Finds a field group based on the specified parameters.
list

Finds all field groups based on the specified parameters.
create

Creates a new field group.

Examples:

Get a field group by reference

>>> api.field_groups.get('_mytenant.field_groups.7a5416d13572')
<FieldGroup 7a5416d13572 title="My Field Group" version="1.0">

Find a field group by title

>>> api.tenant_field_groups.find(title='My Field Group')
<FieldGroup 7a5416d13572 title="My Field Group" version="1.0">

List all global and tenant field groups

>>> api.field_groups.list()
[<FieldGroup xdm.context.identitymap>, <FieldGroup 7a5416d13572>, ...]

List all global field groups

>>> api.global_field_groups.list()
[<FieldGroup xdm.context.identitymap>, <FieldGroup xdm.mixins.current-weather>, ...]

List all tenant field groups

>>> api.tenant_field_groups.list()
[<FieldGroup 7a5416d13572>, <FieldGroup 7a5416d13571>, ...]
Source code in aezpz/schema.py 
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
class FieldGroupCollection(ResourceCollection):
    """ Collection of FieldGroup resources.

    Initialized through `api.field_groups`, `api.global_field_groups`, or `api.tenant_field_groups`.

    Methods:
        get: Retrieves a field group based on the provided reference.
        find: Finds a field group based on the specified parameters.
        list: Finds all field groups based on the specified parameters.
        create: Creates a new field group.

    Examples:
        Get a field group by reference
        >>> api.field_groups.get('_mytenant.field_groups.7a5416d13572')
        <FieldGroup 7a5416d13572 title="My Field Group" version="1.0">

        Find a field group by title
        >>> api.tenant_field_groups.find(title='My Field Group')
        <FieldGroup 7a5416d13572 title="My Field Group" version="1.0">

        List all global and tenant field groups
        >>> api.field_groups.list()
        [<FieldGroup xdm.context.identitymap>, <FieldGroup 7a5416d13572>, ...]

        List all global field groups
        >>> api.global_field_groups.list()
        [<FieldGroup xdm.context.identitymap>, <FieldGroup xdm.mixins.current-weather>, ...]

        List all tenant field groups
        >>> api.tenant_field_groups.list()
        [<FieldGroup 7a5416d13572>, <FieldGroup 7a5416d13571>, ...]
    """
    def __init__(self, api: Api, container:Optional[Container]=None):
        super().__init__(api, container=container, resources=[ResourceType.FIELD_GROUP])

    def get(self, id) -> FieldGroup:
        return super().get(id)

    def find(self, full:bool=True, **params) -> FieldGroup:
        return super().find(full, **params)

    def list(self, full:bool=False, **params) -> list[FieldGroup]:
        return super().list(full, **params)

    def create(self,
               title: str,
               description: str = '',
               properties: dict[str, dict] = {},
               intendedToExtend: list[Resource] = [],
               ) -> FieldGroup:
        """
        Create a new field group.

        Args:
            title: The title of the field group.
            description: The description of the field group.
            properties: The properties of the field group.
            intendedToExtend: The resources this field group intends to extend.

        Returns:
            FieldGroup: The created field group.

        Examples:
            >>> field_group = api.field_groups.create(
            ...     title='My Field Group',
            ...     description='My test field group',
            ...     properties={
            ...         '_mytenant': {
            ...             'type': 'object',
            ...             'properties': {
            ...                 'is_super_star': {'type': 'boolean'},
            ...             }
            ...         },
            ...     }
            ...     intendedToExtend=[api.classes.get('_xdm.context.profile')],
            ... )
        """
        for r in intendedToExtend:
            assert isinstance(r, Resource)
        return self._create({
            'type': 'object',
            'title': title,
            'description': description,
            'meta:intendedToExtend': [ ctx.ref for ctx in intendedToExtend ],
            'allOf': [{
                'properties': properties,
            }]
        })

create(title, description='', properties={}, intendedToExtend=[])

Create a new field group.

Parameters:

Name Type Description Default
title str

The title of the field group.

 required
description str

The description of the field group.

 ''
properties dict[str, dict]

The properties of the field group.

 {}
intendedToExtend list[Resource]

The resources this field group intends to extend.

 []

Returns:

Name Type Description
FieldGroup FieldGroup

The created field group.

Examples:

>>> field_group = api.field_groups.create(
...     title='My Field Group',
...     description='My test field group',
...     properties={
...         '_mytenant': {
...             'type': 'object',
...             'properties': {
...                 'is_super_star': {'type': 'boolean'},
...             }
...         },
...     }
...     intendedToExtend=[api.classes.get('_xdm.context.profile')],
... )
Source code in aezpz/schema.py 
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
def create(self,
           title: str,
           description: str = '',
           properties: dict[str, dict] = {},
           intendedToExtend: list[Resource] = [],
           ) -> FieldGroup:
    """
    Create a new field group.

    Args:
        title: The title of the field group.
        description: The description of the field group.
        properties: The properties of the field group.
        intendedToExtend: The resources this field group intends to extend.

    Returns:
        FieldGroup: The created field group.

    Examples:
        >>> field_group = api.field_groups.create(
        ...     title='My Field Group',
        ...     description='My test field group',
        ...     properties={
        ...         '_mytenant': {
        ...             'type': 'object',
        ...             'properties': {
        ...                 'is_super_star': {'type': 'boolean'},
        ...             }
        ...         },
        ...     }
        ...     intendedToExtend=[api.classes.get('_xdm.context.profile')],
        ... )
    """
    for r in intendedToExtend:
        assert isinstance(r, Resource)
    return self._create({
        'type': 'object',
        'title': title,
        'description': description,
        'meta:intendedToExtend': [ ctx.ref for ctx in intendedToExtend ],
        'allOf': [{
            'properties': properties,
        }]
    })

DataTypeCollection

Bases: ResourceCollection

Collection of DataType resources.

Initialized through api.data_types, api.global_data_types, or api.tenant_data_types.

Methods:

Name Description
get

Retrieves a data type based on the provided reference.
find

Finds a data type based on the specified parameters.
list

Finds all data types based on the specified parameters.
create

Creates a new data type.

Examples:

Get a data type by reference

>>> api.data_types.get('_mytenant.data_types.7a5416d13572')
<DataType 7a5416d13572 title="My Data Type" version="1.0">

Find a data type by title

>>> api.tenant_data_types.find(title='My Data Type')
<DataType 7a5416d13572 title="My Data Type" version="1.0">

List all global and tenant data types

>>> api.data_types.list()
[<DataType xdm.context.person>, <DataType 7a5416d13572>, ...]

List all global data types

>>> api.global_data_types.list()
[<DataType xdm.context.person>, <DataType xdm.context.person-name>, ...]

List all tenant data types

>>> api.tenant_data_types.list()
[<DataType 7a5416d13572>, <DataType 7a5416d13571>, ...]
Source code in aezpz/schema.py 
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
class DataTypeCollection(ResourceCollection):
    """ Collection of DataType resources.

    Initialized through `api.data_types`, `api.global_data_types`, or `api.tenant_data_types`.

    Methods:
        get: Retrieves a data type based on the provided reference.
        find: Finds a data type based on the specified parameters.
        list: Finds all data types based on the specified parameters.
        create: Creates a new data type.

    Examples:
        Get a data type by reference
        >>> api.data_types.get('_mytenant.data_types.7a5416d13572')
        <DataType 7a5416d13572 title="My Data Type" version="1.0">

        Find a data type by title
        >>> api.tenant_data_types.find(title='My Data Type')
        <DataType 7a5416d13572 title="My Data Type" version="1.0">

        List all global and tenant data types
        >>> api.data_types.list()
        [<DataType xdm.context.person>, <DataType 7a5416d13572>, ...]

        List all global data types
        >>> api.global_data_types.list()
        [<DataType xdm.context.person>, <DataType xdm.context.person-name>, ...]

        List all tenant data types
        >>> api.tenant_data_types.list()
        [<DataType 7a5416d13572>, <DataType 7a5416d13571>, ...]
    """
    def __init__(self, api: Api, container:Optional[Container]=None):
        super().__init__(api, container=container, resources=[ResourceType.DATA_TYPE])

    def get(self, id) -> DataType:
        return super().get(id)

    def find(self, full:bool=True, **params) -> DataType:
        return super().find(full, **params)

    def list(self, full:bool=False, **params) -> list[DataType]:
        return super().list(full, **params)

    def create(self,
               title: str,
               description: str = '',
               properties: dict[str, dict] = {},
               ) -> DataType:
        """
        Create a new data type.

        Args:
            title (str): The title of the data type.
            description (str, optional): The description of the data type. Defaults to ''.
            properties (dict[str, dict], optional): The properties of the data type. Defaults to {}.

        Returns:
            DataType: The created data type.

        Examples:
            >>> data_type = api.data_types.create(
            ...     title='My Data Type',
            ...     description='My test data type',
            ...     properties={
            ...         '_mytenant': {
            ...             'type': 'object',
            ...             'properties': {
            ...                 'is_super_star': {'type': 'boolean'},
            ...             }
            ...         },
            ...     }
            ... )
        """
        return self._create({
            'type': 'object',
            'title': title,
            'description': description,
            'properties': properties,
        })

create(title, description='', properties={})

Create a new data type.

Parameters:

Name Type Description Default
title str

The title of the data type.

 required
description str

The description of the data type. Defaults to ''.

 ''
properties dict[str, dict]

The properties of the data type. Defaults to {}.

 {}

Returns:

Name Type Description
DataType DataType

The created data type.

Examples:

>>> data_type = api.data_types.create(
...     title='My Data Type',
...     description='My test data type',
...     properties={
...         '_mytenant': {
...             'type': 'object',
...             'properties': {
...                 'is_super_star': {'type': 'boolean'},
...             }
...         },
...     }
... )
Source code in aezpz/schema.py 
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
def create(self,
           title: str,
           description: str = '',
           properties: dict[str, dict] = {},
           ) -> DataType:
    """
    Create a new data type.

    Args:
        title (str): The title of the data type.
        description (str, optional): The description of the data type. Defaults to ''.
        properties (dict[str, dict], optional): The properties of the data type. Defaults to {}.

    Returns:
        DataType: The created data type.

    Examples:
        >>> data_type = api.data_types.create(
        ...     title='My Data Type',
        ...     description='My test data type',
        ...     properties={
        ...         '_mytenant': {
        ...             'type': 'object',
        ...             'properties': {
        ...                 'is_super_star': {'type': 'boolean'},
        ...             }
        ...         },
        ...     }
        ... )
    """
    return self._create({
        'type': 'object',
        'title': title,
        'description': description,
        'properties': properties,
    })

BehaviorCollection

Bases: ResourceCollection

Collection of Behavior resources.

Initialized through api.behaviors.

Attributes:

Name Type Description
adhoc Behavior

The adhoc behavior.
record Behavior

The record behavior.
time_series Behavior

The time series behavior.

Examples:

>>> api.behaviors.adhoc
<Behavior xdm.data.adhoc>

>>> api.behaviors.record
<Behavior xdm.data.record>

>>> api.behaviors.time_series
<Behavior xdm.data.time-series>
Source code in aezpz/schema.py 
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
class BehaviorCollection(ResourceCollection):
    """ Collection of Behavior resources.

    Initialized through `api.behaviors`.

    Attributes:
        adhoc: The adhoc behavior.
        record: The record behavior.
        time_series: The time series behavior.

    Examples:
        >>> api.behaviors.adhoc
        <Behavior xdm.data.adhoc>

        >>> api.behaviors.record
        <Behavior xdm.data.record>

        >>> api.behaviors.time_series
        <Behavior xdm.data.time-series>
    """

    adhoc: Behavior
    record: Behavior
    time_series: Behavior

    def __init__(self, api: Api, container:Optional[Container]=None):
        super().__init__(api, container=container, resources=[ResourceType.BEHAVIOR])
        self.adhoc = Behavior(self.api, 'https://ns.adobe.com/xdm/data/adhoc')
        self.record = Behavior(self.api, 'https://ns.adobe.com/xdm/data/record')
        self.time_series = Behavior(self.api, 'https://ns.adobe.com/xdm/data/time-series')

    def get(self, id) -> Behavior:
        return super().get(id)

    def find(self, full:bool=True, **params) -> Behavior:
        return super().find(full, **params)

    def list(self, full:bool=False, **params) -> list[Behavior]:
        return super().list(full, **params)

Resource

Base class for all resources.

Attributes:

Name Type Description
body dict

The raw body of the resource.
id str

The meta:altId of the resource.
ref str

The $id of the resource.
uuid str

The unique identifier part of the id or ref.
container Container

The container of the resource either "global" or "tenant".
tenant Optional[str]

The tenant name used in the resource id. Available only for tenant resources.
version str

The version of the resource.
title str

The title of the resource.
description str

The description of the resource.
extends list[Resource]

The list of resources that the resource extends.

Methods:

Name Description
get

Refreshes the data to be in sync with the server.
delete

Deletes the resource.

Examples:

>>> schema = api.schemas.get('_mytenant.schemas.7a5416d13572')
<Schema 7a5416d13572 title="My Schema" version="1.0">

Update the title of a schema (will send a PATCH request to the server)

>>> schema.title = 'My New Schema'

Update the description of a schema

>>> schema.description = 'My new test schema'

Get attribute from the raw response body

>>> schema.body['meta:altId']
'_mytenant.schemas.7a5416d13572'

Delete the schema

>>> schema.delete()
Source code in aezpz/schema.py 
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
class Resource:
    """ Base class for all resources.

    Attributes:
        body: The raw body of the resource.
        id: The `meta:altId` of the resource.
        ref: The `$id` of the resource.
        uuid: The unique identifier part of the id or ref.
        container: The container of the resource either "global" or "tenant".
        tenant: The tenant name used in the resource id. Available only for tenant resources.

        version: The version of the resource.
        title: The title of the resource.
        description: The description of the resource.
        extends: The list of resources that the resource extends.

    Methods:
        get: Refreshes the data to be in sync with the server.
        delete: Deletes the resource.

    Examples:
        >>> schema = api.schemas.get('_mytenant.schemas.7a5416d13572')
        <Schema 7a5416d13572 title="My Schema" version="1.0">

        Update the title of a schema (will send a PATCH request to the server)
        >>> schema.title = 'My New Schema'

        Update the description of a schema
        >>> schema.description = 'My new test schema'

        Get attribute from the raw response body
        >>> schema.body['meta:altId']
        '_mytenant.schemas.7a5416d13572'

        Delete the schema
        >>> schema.delete()
    """

    api: Api
    body: dict
    type: ResourceType

    id: str
    ref: str
    uuid: str
    tenant: Optional[str]
    container: Container

    def __init__(self, api: Api, body):
        if type(body) is str:
            ref = SchemaRef(body)
            body = { '$id': ref.ref }
        elif isinstance(body, SchemaRef):
            ref = body
            body = { '$id': ref.ref }
        elif type(body) is dict:
            ref = SchemaRef(body['$id'])
        else:
            raise TypeError(f'Unexpected body type: {body}')

        self.api = api
        self.body = body
        self.id = ref.id
        self.uuid = ref.uuid
        self.ref = ref.ref
        self.tenant = ref.tenant
        self.container = ref.container
        self.type = self.__class__.type
        assert self.type == ref.resource, 'Mismatched resource type'

    def request(self, method, full:bool=False, json=None):
        r = self.api.request(
            method=method,
            path=get_resource_path(
                container=self.container,
                resource=self.type,
                id=self.id,
            ),
            headers=get_accept_header(
                xed='full' if full else None
            ),
            json=json,
        )
        if r is not None:
            assert r['$id'] == self.ref
            assert r['meta:altId'] == self.id
            self.body.update(**r)
        return self

    @property
    def version(self) -> str:
        if 'version' not in self.body:
            self.get(full=False)
        return self.body['version']

    @property
    def title(self) -> str:
        if 'title' not in self.body:
            self.get(full=False)
        return self.body['title']

    @title.setter
    def title(self, value):
        self.request('PATCH', json=[{
            'op': 'replace',
            'path': '/title',
            'value': value,
        }])

    @property
    def description(self) -> str:
        if 'description' not in self.body:
            self.get(full=False)
        return self.body['description']

    @description.setter
    def description(self, value):
        self.request('PATCH', json=[{
            'op': 'replace',
            'path': '/description',
            'value': value,
        }])

    @property
    def properties(self) -> dict[str, dict]:
        if 'properties' not in self.body:
            self.get(full=True)
        self.body.setdefault('properties', {})
        return self.body['properties']

    @property
    def definitions(self) -> dict[str, dict]:
        if 'allOf' not in self.body:
            self.get(full=False)
        self.body.setdefault('allOf', [])
        properties = {}
        for record in self.body['allOf']:
            definition = record
            if record.get('$ref','').startswith('#'):
                assert 'properties' not in record, 'unexpected "properties" and "$ref" definition'
                assert record['$ref'].startswith('#/definitions/'), 'unexpected non-definitions reference'
                field = record['$ref'][len('#/definitions/'):]
                assert '/' not in field, 'unexpected nested definition reference'
                definition = self.body.get('definitions',{}).get(field)
                assert definition is not None, 'reference to missing definition'
                assert 'properties' in definition, 'expected definition to be an object'

            for key,val in definition.get('properties',{}).items():
                assert key not in properties, 'unhandled merging of definitions'
                properties[key] = val
        return properties

    @property
    def extends(self) -> list[Resource]:
        if 'meta:extends' not in self.body:
            self.get(full=False)
        self.body.setdefault('meta:extends', [])
        extends = []
        for ref in self.body['meta:extends']:
            extends.append(SchemaRef(ref).init(self.api))
        return extends

    def get(self, full=True):
        return self.request('GET', full=full)

    def delete(self):
        self.request('DELETE')

    def __repr__(self):
        return '<{class_name} {uuid}{title}{version}>'.format(
            class_name=self.__class__.__name__,
            uuid=self.uuid,
            title=f' title="{self.title}"' if 'title' in self.body else '',
            version=f' version="{self.version}"' if 'version' in self.body else '',
        )

Schema

Bases: Resource

A schema resource.

Attributes:

Name Type Description
parent Class

The parent class that the schema inherits from.
behavior Behavior

The behavior of the schema.
field_groups list[FieldGroup]

The list of field groups used in the schema.
Source code in aezpz/schema.py 
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
class Schema(Resource):
    """ A schema resource.

    Attributes:
        parent: The parent class that the schema inherits from.
        behavior: The behavior of the schema.
        field_groups: The list of field groups used in the schema.
    """

    type = ResourceType.SCHEMA

    @property
    def parent(self) -> Class:
        if 'meta:class' not in self.body:
            self.get()
        return Class(self.api, self.body['meta:class'])

    @property
    def behavior(self) -> Behavior:
        behaviors = [r for r in self.extends if r.type == ResourceType.BEHAVIOR]
        assert len(behaviors) == 1
        return behaviors[0]

    @property
    def field_groups(self) -> list[FieldGroup]:
        return [resource for resource in self.extends if resource.type == ResourceType.FIELD_GROUP]

    def add_field_group(self, field_group: FieldGroup):
        assert isinstance(field_group, FieldGroup)
        r = self.request('PATCH', json=[
            { 'op': 'add', 'path': '/allOf/-', 'value': {'$ref': field_group.id} }
        ])
        return self(**r)

Class

Bases: Resource

A class resource.

Attributes:

Name Type Description
behavior Behavior

The behavior of the class.
field_groups list[FieldGroup]

The list of field groups used in the class.
Source code in aezpz/schema.py 
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
class Class(Resource):
    """ A class resource.

    Attributes:
        behavior: The behavior of the class.
        field_groups: The list of field groups used in the class.
    """
    type = ResourceType.CLASS

    @property
    def behavior(self) -> Behavior:
        behaviors = [r for r in self.extends if r.type == ResourceType.BEHAVIOR]
        assert len(behaviors) == 1
        return behaviors[0]

    @property
    def field_groups(self) -> list[FieldGroup]:
        return [resource for resource in self.extends if resource.type == ResourceType.FIELD_GROUP]

FieldGroup

Bases: Resource

A field group resource.

Attributes:

Name Type Description
intendedToExtend

The resources this field group intends to extend.
Source code in aezpz/schema.py 
869
870
871
872
873
874
875
876
877
878
879
880
881
882
class FieldGroup(Resource):
    """ A field group resource.

    Attributes:
        intendedToExtend: The resources this field group intends to extend.
    """
    type = ResourceType.FIELD_GROUP

    @property
    def intendedToExtend(self):
        if 'meta:intendedToExtend' not in self.body:
            self.get(full=False)
        self.body.setdefault('meta:intendedToExtend', [])
        return [SchemaRef(ref).init(self.api) for ref in self.body['meta:intendedToExtend']]

DataType

Bases: Resource

A data type resource.

Source code in aezpz/schema.py 
884
885
886
887
class DataType(Resource):
    """ A data type resource.
    """
    type = ResourceType.DATA_TYPE

Behavior

Bases: Resource

A behavior resource.

Source code in aezpz/schema.py 
889
890
891
892
class Behavior(Resource):
    """ A behavior resource.
    """
    type = ResourceType.BEHAVIOR

DatasetCollection

Collection of Datasets.

Initialized through api.datasets.

Methods:

Name Description
get

Find an existing dataset by id.
find

Find an existing dataset by name.
list

List all datasets.
create

Create a new dataset.
Source code in aezpz/datasets.py 
 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
class DatasetCollection:
    """ Collection of Datasets.

    Initialized through `api.datasets`.

    Methods:
        get: Find an existing dataset by id.
        find: Find an existing dataset by name.
        list: List all datasets.
        create: Create a new dataset.
    """
    api: Api

    def __init__(self, api: Api):
        self.api = api

    def get(self, id) -> DataSet:
        """ Get a dataset by id.

        Examples:
            >>> api.datasets.get('65cfb1ca9e91f22')
            <DataSet 65cfb1ca9e91f22>
        """
        return DataSet(api=self.api, id=id)

    def find(self, **kwargs) -> DataSet:
        """ Find a dataset by properties

        Examples:
            >>> api.datasets.find(name='My Dataset')
            <DataSet 65cfb1ca9e91f22 name="My Dataset" version="1.0.0">
        """
        datasets = self.list(**kwargs)
        assert len(datasets) == 1, f"Expected to find exactly one dataset, but found {len(datasets)}"
        return datasets[0]

    def list(self, **kwargs) -> list[DataSet]:
        """ List all datasets that match the given properties.

        Examples:
            >>> api.datasets.list(name="untitled")
            [<DataSet 65cfb1ca9e91f22 name="untitled" version="1.0.0">, ...]
        """
        params = kwargs
        datasets = []
        while True:
            r = self.api.request('GET', '/data/foundation/catalog/dataSets', params=params)
            for k, v in r.items():
                datasets.append(DataSet(self.api, k, v))
            if len(r) == 100:
                params['start'] = params.get('start',0) + 100
            else:
                break
        return datasets

    def create(self,
               schema: Schema,
               name: str = None,
               description: str = None,
               tags: dict[str, Union[str, list[str]]] = None,
               enableErrorDiagnostics: bool = None,
               observableSchema: dict = None,
               format: Literal['','csv','text','json','parquet','sequencefile','avro'] = None,
               delimiters: Union[str, list[str]] = None,
               quotes: Union[str, list[str]] = None,
               escapes: Union[str, list[str]] = None,
               nullMarkers: Union[str, list[str]] = None,
               header: bool = None,
               charset: Literal['US-ASCII','UTF-8','ISO-8859-1',''] = None,
               dataSourceId: int = None,
              ) -> DataSet:
        """ Create a new dataset.

        Args:
            schema: The schema of the dataset.
            name: A descriptive, human-readable name for this dataset.
            description: A longer text description of the dataset.
            tags: Tags are values associated with a particular object, 
                these are generally used by external systems for marking an object 
                in a way that it understands. Normally tags are not used for 
                internal Catalog business logic
            enableErrorDiagnostics: This field provides the ability to opt in to 
                generating diagnostic files for the errors while ingesting data.
            observableSchema: observableSchema stores a JSON Schema object that is a 
                valid subset (including the null case) of the XDM model specified by schemaRef.
            format: The file format for all dataSetFiles associated with this view. 
                Required for CSV upload workflows, but optional in all other cases.
            delimiters: Characters used to separate fields for the file format.
            quotes: Quote characters used for the file format.
            escapes: Escape characters used for the file format.
            nullMarkers: Null markers used for the file format.
            header: Whether the file format has a header.
            charset: The character encoding of the files..
            dataSourceId: The ID of the datasource. It must be unique, along with its name.

        Examples:
            >>> api.datasets.create()
            <DataSet 65cfb1ca9e91f22 name="untitled" version="1.0.0">
        """
        body = form_dataset_body(
            name=name,
            description=description,
            tags=tags,
            enableErrorDiagnostics=enableErrorDiagnostics,
            observableSchema=observableSchema,
            format=format,
            delimiters=delimiters,
            quotes=quotes,
            escapes=escapes,
            nullMarkers=nullMarkers,
            header=header,
            charset=charset,
            schema=schema,
            dataSourceId=dataSourceId,
        )
        r = self.api.request('POST', '/data/foundation/catalog/dataSets', json=body)
        return DataSet(self.api, parse_id_list(r))

get(id)

Get a dataset by id.

Examples:

>>> api.datasets.get('65cfb1ca9e91f22')
<DataSet 65cfb1ca9e91f22>
Source code in aezpz/datasets.py 
88
89
90
91
92
93
94
95
def get(self, id) -> DataSet:
    """ Get a dataset by id.

    Examples:
        >>> api.datasets.get('65cfb1ca9e91f22')
        <DataSet 65cfb1ca9e91f22>
    """
    return DataSet(api=self.api, id=id)

find(**kwargs)

Find a dataset by properties

Examples:

>>> api.datasets.find(name='My Dataset')
<DataSet 65cfb1ca9e91f22 name="My Dataset" version="1.0.0">
Source code in aezpz/datasets.py 
 97
 98
 99
100
101
102
103
104
105
106
def find(self, **kwargs) -> DataSet:
    """ Find a dataset by properties

    Examples:
        >>> api.datasets.find(name='My Dataset')
        <DataSet 65cfb1ca9e91f22 name="My Dataset" version="1.0.0">
    """
    datasets = self.list(**kwargs)
    assert len(datasets) == 1, f"Expected to find exactly one dataset, but found {len(datasets)}"
    return datasets[0]

list(**kwargs)

List all datasets that match the given properties.

Examples:

>>> api.datasets.list(name="untitled")
[<DataSet 65cfb1ca9e91f22 name="untitled" version="1.0.0">, ...]
Source code in aezpz/datasets.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def list(self, **kwargs) -> list[DataSet]:
    """ List all datasets that match the given properties.

    Examples:
        >>> api.datasets.list(name="untitled")
        [<DataSet 65cfb1ca9e91f22 name="untitled" version="1.0.0">, ...]
    """
    params = kwargs
    datasets = []
    while True:
        r = self.api.request('GET', '/data/foundation/catalog/dataSets', params=params)
        for k, v in r.items():
            datasets.append(DataSet(self.api, k, v))
        if len(r) == 100:
            params['start'] = params.get('start',0) + 100
        else:
            break
    return datasets

create(schema, name=None, description=None, tags=None, enableErrorDiagnostics=None, observableSchema=None, format=None, delimiters=None, quotes=None, escapes=None, nullMarkers=None, header=None, charset=None, dataSourceId=None)

Create a new dataset.

Parameters:

Name Type Description Default
schema Schema

The schema of the dataset.

 required
name str

A descriptive, human-readable name for this dataset.

 None
description str

A longer text description of the dataset.

 None
tags dict[str, Union[str, list[str]]]

Tags are values associated with a particular object, these are generally used by external systems for marking an object in a way that it understands. Normally tags are not used for internal Catalog business logic

 None
enableErrorDiagnostics bool

This field provides the ability to opt in to generating diagnostic files for the errors while ingesting data.

 None
observableSchema dict

observableSchema stores a JSON Schema object that is a valid subset (including the null case) of the XDM model specified by schemaRef.

 None
format Literal['', 'csv', 'text', 'json', 'parquet', 'sequencefile', 'avro']

The file format for all dataSetFiles associated with this view. Required for CSV upload workflows, but optional in all other cases.

 None
delimiters Union[str, list[str]]

Characters used to separate fields for the file format.

 None
quotes Union[str, list[str]]

Quote characters used for the file format.

 None
escapes Union[str, list[str]]

Escape characters used for the file format.

 None
nullMarkers Union[str, list[str]]

Null markers used for the file format.

 None
header bool

Whether the file format has a header.

 None
charset Literal['US-ASCII', 'UTF-8', 'ISO-8859-1', '']

The character encoding of the files..

 None
dataSourceId int

The ID of the datasource. It must be unique, along with its name.

 None

Examples:

>>> api.datasets.create()
<DataSet 65cfb1ca9e91f22 name="untitled" version="1.0.0">
Source code in aezpz/datasets.py 
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
def create(self,
           schema: Schema,
           name: str = None,
           description: str = None,
           tags: dict[str, Union[str, list[str]]] = None,
           enableErrorDiagnostics: bool = None,
           observableSchema: dict = None,
           format: Literal['','csv','text','json','parquet','sequencefile','avro'] = None,
           delimiters: Union[str, list[str]] = None,
           quotes: Union[str, list[str]] = None,
           escapes: Union[str, list[str]] = None,
           nullMarkers: Union[str, list[str]] = None,
           header: bool = None,
           charset: Literal['US-ASCII','UTF-8','ISO-8859-1',''] = None,
           dataSourceId: int = None,
          ) -> DataSet:
    """ Create a new dataset.

    Args:
        schema: The schema of the dataset.
        name: A descriptive, human-readable name for this dataset.
        description: A longer text description of the dataset.
        tags: Tags are values associated with a particular object, 
            these are generally used by external systems for marking an object 
            in a way that it understands. Normally tags are not used for 
            internal Catalog business logic
        enableErrorDiagnostics: This field provides the ability to opt in to 
            generating diagnostic files for the errors while ingesting data.
        observableSchema: observableSchema stores a JSON Schema object that is a 
            valid subset (including the null case) of the XDM model specified by schemaRef.
        format: The file format for all dataSetFiles associated with this view. 
            Required for CSV upload workflows, but optional in all other cases.
        delimiters: Characters used to separate fields for the file format.
        quotes: Quote characters used for the file format.
        escapes: Escape characters used for the file format.
        nullMarkers: Null markers used for the file format.
        header: Whether the file format has a header.
        charset: The character encoding of the files..
        dataSourceId: The ID of the datasource. It must be unique, along with its name.

    Examples:
        >>> api.datasets.create()
        <DataSet 65cfb1ca9e91f22 name="untitled" version="1.0.0">
    """
    body = form_dataset_body(
        name=name,
        description=description,
        tags=tags,
        enableErrorDiagnostics=enableErrorDiagnostics,
        observableSchema=observableSchema,
        format=format,
        delimiters=delimiters,
        quotes=quotes,
        escapes=escapes,
        nullMarkers=nullMarkers,
        header=header,
        charset=charset,
        schema=schema,
        dataSourceId=dataSourceId,
    )
    r = self.api.request('POST', '/data/foundation/catalog/dataSets', json=body)
    return DataSet(self.api, parse_id_list(r))

DataSet

Source code in aezpz/datasets.py 
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
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
class DataSet:
    api: Api
    id: str
    _body: dict

    def __init__(self, api: Api, id: str, body: dict = None):
        self.api = api
        self.id = id
        self._body = body

    def request(self, method: Literal['GET', 'POST', 'PUT', 'DELETE'], json: Optional[dict[str, Any]] = None):
        return self.api.request(method, f'/data/foundation/catalog/dataSets/{self.id}', json=json)

    @property
    def body(self):
        if self._body is None:
            self.get()
        return self._body

    @property
    def name(self) -> str:
        return self.body.get('name')

    @property
    def description(self) -> str:
        return self.body.get('description')

    @property
    def tags(self) -> dict[str, list[str]]:
        return self.body.get('tags')

    @property
    def enableErrorDiagnostics(self) -> bool:
        return self.body.get('enableErrorDiagnostics')

    @property
    def observableSchema(self) -> dict:
        return self.body.get('observableSchema')

    @property
    def format(self) -> Literal['','csv','text','json','parquet','sequencefile','avro']:
        return self.body.get('fileDescription', {}).get('format')

    @property
    def delimiters(self) -> list[str]:
        return self.body.get('fileDescription', {}).get('delimiters', [])

    @property
    def quotes(self) -> list[str]:
        return self.body.get('fileDescription', {}).get('quotes', [])

    @property
    def escapes(self) -> list[str]:
        return self.body.get('fileDescription', {}).get('escapes', [])

    @property
    def nullMarkers(self) -> list[str]:
        return self.body.get('fileDescription', {}).get('nullMarkers', [])

    @property
    def header(self) -> bool:
        return self.body.get('fileDescription', {}).get('header')

    @property
    def charset(self) -> Literal['US-ASCII','UTF-8','ISO-8859-1','']:
        return self.body.get('fileDescription', {}).get('charset', '')

    @property
    def schema(self):
        if 'schemaRef' not in self.body:
            return None
        return self.api.schemas.get(self.body.get('schemaRef')['id'])

    @property
    def dataSourceId(self) -> int:
        return self.body.get('dataSourceId')

    @property
    def version(self) -> str:
        return self.body.get('version')

    @property
    def created(self) -> datetime.datetime:
        return datetime.datetime.fromtimestamp(self.body.get('created') / 1000)

    @property
    def updated(self) -> datetime.datetime:
        return datetime.datetime.fromtimestamp(self.body.get('updated') / 1000)

    def batches(self) -> list[Batch]:
        return self.api.batches.list(dataSet=self.id)

    def upload(self,
               filepath: str,
               format: Literal['json','jsonl','parquet','csv'],
               replace: Union[bool, list[Batch]] = []
            ) -> Batch:
        batch = self.api.batches.create(dataset=self, format=format, replace=replace)
        batch.upload(filepath)
        batch.complete()
        return self

    def get(self) -> DataSet:
        r = self.request('GET')
        assert self.id in r
        if self._body is None:
            self._body = {}
        self._body.update(**r[self.id])
        return self

    def update(self,
                name: str = None,
                description: str = None,
                tags: dict[str, Union[str, list[str]]] = None,
                enableErrorDiagnostics: bool = None,
                observableSchema: dict = None,
                format: Literal['','csv','text','json','parquet','sequencefile','avro'] = None,
                delimiters: Union[str, list[str]] = None,
                quotes: Union[str, list[str]] = None,
                escapes: Union[str, list[str]] = None,
                nullMarkers: Union[str, list[str]] = None,
                header: bool = None,
                charset: Literal['US-ASCII','UTF-8','ISO-8859-1',''] = None,
                schema: Schema = None,
                dataSourceId: int = None,
            ) -> DataSet:
        """ Update the dataset.

        Args:
            name: A descriptive, human-readable name for this dataset.
            description: A longer text description of the dataset.
            tags: Tags are values associated with a particular object, 
                these are generally used by external systems for marking an object 
                in a way that it understands. Normally tags are not used for 
                internal Catalog business logic
            enableErrorDiagnostics: This field provides the ability to opt in to 
                generating diagnostic files for the errors while ingesting data.
            observableSchema: observableSchema stores a JSON Schema object that is a 
                valid subset (including the null case) of the XDM model specified by schemaRef.
            format: The file format for all dataSetFiles associated with this view. 
                Required for CSV upload workflows, but optional in all other cases.
            delimiters: Characters used to separate fields for the file format.
            quotes: Quote characters used for the file format.
            escapes: Escape characters used for the file format.
            nullMarkers: Null markers used for the file format.
            header: Whether the file format has a header.
            charset: The character encoding of the files..
            schema: The schema of the dataset.
            dataSourceId: The ID of the datasource. It must be unique, along with its name.
        """
        body = form_dataset_body(
            name=name,
            description=description,
            tags=tags,
            enableErrorDiagnostics=enableErrorDiagnostics,
            observableSchema=observableSchema,
            format=format,
            delimiters=delimiters,
            quotes=quotes,
            escapes=escapes,
            nullMarkers=nullMarkers,
            header=header,
            charset=charset,
            schema=schema,
            dataSourceId=dataSourceId,
        )
        self.request('PATCH', json=body)
        self.get()
        return self

    def delete(self) -> DataSet:
        self.request('DELETE')
        return self

    def __repr__(self):
        return '<{class_name} {id}{name}{version}>'.format(
            class_name=self.__class__.__name__,
            id=self.id,
            name=f' name="{self.name}"' if self._body is not None else '',
            version=f' version="{self.version}"' if self._body is not None else '',
        )

update(name=None, description=None, tags=None, enableErrorDiagnostics=None, observableSchema=None, format=None, delimiters=None, quotes=None, escapes=None, nullMarkers=None, header=None, charset=None, schema=None, dataSourceId=None)

Update the dataset.

Parameters:

Name Type Description Default
name str

A descriptive, human-readable name for this dataset.

 None
description str

A longer text description of the dataset.

 None
tags dict[str, Union[str, list[str]]]

Tags are values associated with a particular object, these are generally used by external systems for marking an object in a way that it understands. Normally tags are not used for internal Catalog business logic

 None
enableErrorDiagnostics bool

This field provides the ability to opt in to generating diagnostic files for the errors while ingesting data.

 None
observableSchema dict

observableSchema stores a JSON Schema object that is a valid subset (including the null case) of the XDM model specified by schemaRef.

 None
format Literal['', 'csv', 'text', 'json', 'parquet', 'sequencefile', 'avro']

The file format for all dataSetFiles associated with this view. Required for CSV upload workflows, but optional in all other cases.

 None
delimiters Union[str, list[str]]

Characters used to separate fields for the file format.

 None
quotes Union[str, list[str]]

Quote characters used for the file format.

 None
escapes Union[str, list[str]]

Escape characters used for the file format.

 None
nullMarkers Union[str, list[str]]

Null markers used for the file format.

 None
header bool

Whether the file format has a header.

 None
charset Literal['US-ASCII', 'UTF-8', 'ISO-8859-1', '']

The character encoding of the files..

 None
schema Schema

The schema of the dataset.

 None
dataSourceId int

The ID of the datasource. It must be unique, along with its name.

 None
Source code in aezpz/datasets.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
def update(self,
            name: str = None,
            description: str = None,
            tags: dict[str, Union[str, list[str]]] = None,
            enableErrorDiagnostics: bool = None,
            observableSchema: dict = None,
            format: Literal['','csv','text','json','parquet','sequencefile','avro'] = None,
            delimiters: Union[str, list[str]] = None,
            quotes: Union[str, list[str]] = None,
            escapes: Union[str, list[str]] = None,
            nullMarkers: Union[str, list[str]] = None,
            header: bool = None,
            charset: Literal['US-ASCII','UTF-8','ISO-8859-1',''] = None,
            schema: Schema = None,
            dataSourceId: int = None,
        ) -> DataSet:
    """ Update the dataset.

    Args:
        name: A descriptive, human-readable name for this dataset.
        description: A longer text description of the dataset.
        tags: Tags are values associated with a particular object, 
            these are generally used by external systems for marking an object 
            in a way that it understands. Normally tags are not used for 
            internal Catalog business logic
        enableErrorDiagnostics: This field provides the ability to opt in to 
            generating diagnostic files for the errors while ingesting data.
        observableSchema: observableSchema stores a JSON Schema object that is a 
            valid subset (including the null case) of the XDM model specified by schemaRef.
        format: The file format for all dataSetFiles associated with this view. 
            Required for CSV upload workflows, but optional in all other cases.
        delimiters: Characters used to separate fields for the file format.
        quotes: Quote characters used for the file format.
        escapes: Escape characters used for the file format.
        nullMarkers: Null markers used for the file format.
        header: Whether the file format has a header.
        charset: The character encoding of the files..
        schema: The schema of the dataset.
        dataSourceId: The ID of the datasource. It must be unique, along with its name.
    """
    body = form_dataset_body(
        name=name,
        description=description,
        tags=tags,
        enableErrorDiagnostics=enableErrorDiagnostics,
        observableSchema=observableSchema,
        format=format,
        delimiters=delimiters,
        quotes=quotes,
        escapes=escapes,
        nullMarkers=nullMarkers,
        header=header,
        charset=charset,
        schema=schema,
        dataSourceId=dataSourceId,
    )
    self.request('PATCH', json=body)
    self.get()
    return self