Skip to content

LumApps Client

__init__(self, customer_id, instance_id, *args, *, cache=None, dry_run=False, **kwargs) special ¤

Create a LumAppsClient associated to a particular LumApps platform and site

Parameters:

Name Type Description Default
customer_id str

The id of the platform you target

required
instance_id Optional[str]

The id of the instance you target

required
args

The args to pass to the BaseClient

()
cache Optional[Any]

The cache to use

None
dry_run bool

Whether to run in dry_run mode or not. This will avoid saving things when callings save endpoints

False
kwargs

The kwargs to pass to the BaseClient

{}
Source code in lumapps/api/client.py
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
def __init__(
    self,
    customer_id: str,
    instance_id: Optional[str],
    *args,
    cache: Optional[Any] = None,
    dry_run: bool = False,
    **kwargs,
):
    """ Create a LumAppsClient associated to a particular LumApps platform and site

        Args:
            customer_id: The id of the platform you target
            instance_id: The id of the instance you target
            args: The args to pass to the BaseClient
            cache: The cache to use
            dry_run: Whether to run in dry_run mode or not. This will
                avoid saving things when callings save endpoints
            kwargs: The kwargs to pass to the BaseClient
    """
    if not customer_id:
        raise LumAppsClientConfError("customer_id required")
    self.customer_id = customer_id
    self.instance_id = instance_id
    if cache:
        self.cache = cache
    else:
        self.cache = DiscoveryCacheDict()
    self.dry_run = dry_run
    self._langs = None
    super().__init__(*args, **kwargs)
    self._cached_metadata = {}

Content¤

get_available_instance_slug(self, desired_slug) ¤

Find an available instance slug according to the given slug and the current token customer. If the exact slug is not avaialble -i at the end until we found an available one or i reachs 300.

Parameters:

Name Type Description Default
desired_slug str

The desired slug

required

Returns:

Type Description
str

The first available slug found

Exceptions:

Type Description
Exception

Reached 300 try

Source code in lumapps/api/client.py
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
def get_available_instance_slug(self, desired_slug: str) -> str:
    """ Find an available instance slug according to
        the given slug and the current token customer.
        If the exact slug is not avaialble -i at the end until we
        found an available one or i reachs 300.

        Args:
            desired_slug: The desired slug

        Returns:
            The first available slug found

        Raises:
            Exception: Reached 300 try
    """
    post_fix = None
    while True:
        c = self.get_instance(slug=desired_slug, fields="id")
        if not c:
            return desired_slug
        if not post_fix:
            post_fix = 1
            desired_slug += "-1"
        else:
            desired_slug = desired_slug[: -len(str(post_fix)) - 1]
            post_fix += 1
            desired_slug += "-" + str(post_fix)
        if post_fix > 100:
            raise Exception("300 limit as slug postfix")

get_available_slug(self, desired_slug) ¤

Find an available content slug according to the given slug and the current token customer. If the exact slug is not avaialble -i at the end until we found an available one or i reachs 300.

Parameters:

Name Type Description Default
desired_slug str

The desired slug

required

Returns:

Type Description
str

The first available slug found

Exceptions:

Type Description
Exception

Reached 300 try

Source code in lumapps/api/client.py
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
def get_available_slug(self, desired_slug: str) -> str:
    """
        Find an available content slug according to
        the given slug and the current token customer.
        If the exact slug is not avaialble -i at the end until we
        found an available one or i reachs 300.

        Args:
            desired_slug: The desired slug

        Returns:
            The first available slug found

        Raises:
            Exception: Reached 300 try
    """
    post_fix = None
    while True:
        if desired_slug in RESERVED_SLUGS:
            c = True
        else:
            c = self.get_content_by_slug(desired_slug, fields="id")
        if not c:
            return desired_slug
        if not post_fix:
            post_fix = 1
            desired_slug += "-1"
        else:
            desired_slug = desired_slug[: -len(str(post_fix)) - 1]
            post_fix += 1
            desired_slug += "-" + str(post_fix)
        if post_fix > 300:
            raise Exception("300 limit as slug postfix")

get_content(self, content_id, fields=None, action='PAGE_EDIT', cache=False) ¤

Get a content via his id

Parameters:

Name Type Description Default
content_id str

The id of the content to get

required
fields str

The fields projection to apply

None
action str

PAGE_EDIT

'PAGE_EDIT'
cache bool

Whether to cache the result or not

False

Returns:

Type Description
Optional[Dict[str, Any]]

The retrieved content or None if it was not found

Source code in lumapps/api/client.py
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
@none_on_404
def get_content(
    self,
    content_id: str,
    fields: str = None,
    action: str = "PAGE_EDIT",
    cache: bool = False,
) -> Optional[Dict[str, Any]]:
    """ Get a content via his id

        Args:
            content_id: The id of the content to get
            fields: The fields projection to apply
            action: PAGE_EDIT
            cache: Whether to cache the result or not

        Returns:
            The retrieved content or None if it was not found
    """
    if cache:
        c = self.cache.get(f"{self.customer_id}|CONTENT|{content_id}")
        if c:
            return c
    params = {}
    if action:
        params["action"] = action
    if fields:
        params["fields"] = fields
    return self.get_call("content/get", uid=content_id, **params)  # type: ignore

get_content_by_slug(self, slug, fields=None, action='PAGE_EDIT') ¤

Get a content via his slug

Parameters:

Name Type Description Default
slug str

The slug of the content to get

required
fields str

The fields projection to apply

None
action str

PAGE_EDIT

'PAGE_EDIT'

Returns:

Type Description
Optional[Dict[str, Any]]

The retrieved content or None if it was not found

Source code in lumapps/api/client.py
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
@none_on_404
def get_content_by_slug(
    self, slug: str, fields: str = None, action: str = "PAGE_EDIT"
) -> Optional[Dict[str, Any]]:
    """
        Get a content via his slug

        Args:
            slug: The slug of the content to get
            fields: The fields projection to apply
            action: PAGE_EDIT

        Returns:
            The retrieved content or None if it was not found
    """
    params = {}
    if action:
        params["action"] = action
    if fields:
        params["fields"] = fields
    return self.get_call(
        "content/get", instance=self.instance_id, slug=slug, **params
    )  # type: ignore

iter_contents(self, content_type_id=None, **kwargs) ¤

Iterate over the contents on the current lumapps site

https://apiv1.lumapps.com/#operation/Content/List

Parameters:

Name Type Description Default
content_type_id str

The id of a content type. This will be used to filter the retrieved contents

None
kwargs

The args to pass to the request (see lumapps api doc)

{}
Source code in lumapps/api/client.py
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
def iter_contents(
    self, content_type_id: str = None, **kwargs
) -> Generator[Dict[str, Any], None, None]:
    """ Iterate over the contents on the current lumapps site \n

        https://apiv1.lumapps.com/#operation/Content/List

        Args:
            content_type_id: The id of a content type.
                This will be used to filter the retrieved contents
            kwargs: The args to pass to the request (see lumapps api doc)

    """
    body = {"lang": "", "instanceId": self.instance_id, "action": "PAGE_EDIT"}
    if content_type_id:
        body["customContentType"] = content_type_id
    body.update(**kwargs)
    yield from self.iter_call("content/list", body=body)

save_content(self, content, cache=False) ¤

Save a content

Parameters:

Name Type Description Default
content Dict[str, Any]

The content to save

required
cache bool

Whether to cache the saved content based on his id

False

Returns:

Type Description
Dict[str, Any]

The saved content

Source code in lumapps/api/client.py
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
@raise_known_save_errors
def save_content(
    self, content: Dict[str, Any], cache: bool = False
) -> Dict[str, Any]:
    """ Save a content

        Args:
            content: The content to save
            cache: Whether to cache the saved content based on his id

        Returns:
            The saved content
    """
    debug(f"Saving content: {to_json(content)}")
    if self.dry_run:
        return content
    assert not content_is_template(content)
    dst = self.get_call("content/save", body=content, sendNotifications=False)
    if cache:
        self.cache.set(f"{self.customer_id}|CONTENT|{dst['id']}", dst, 5 * 60 * 60)
    return dst

User¤

get_user(self, id_or_email) ¤

Get a user from his id or email

Parameters:

Name Type Description Default
id_or_email str

The id or email or the user

required

Returns:

Type Description
Dict[str, Any]

The retrieved user

Source code in lumapps/api/client.py
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
@lru_cache()
def get_user(self, id_or_email: str) -> Dict[str, Any]:
    """ Get a user from his id or email

        Args:
            id_or_email: The id or email or the user

        Returns:
            The retrieved user
    """
    k = f"{self.customer_id}|USER|{id_or_email}"
    try:
        return self.cache.get(k, raises=True)
    except KeyError:
        pass
    user = self._get_user(id_or_email)
    self.cache.set(k, user, 7200)
    return user

iter_users(self, **kwargs) ¤

Iterate overs the platform users

https://apiv1.lumapps.com/#operation/User/List

Parameters:

Name Type Description Default
**kwargs dict

args to add to the request (see api documentation)

{}

Returns:

Type Description
Generator[Dict[str, Any], NoneType, NoneType]

A generator return the platform users

Source code in lumapps/api/client.py
602
603
604
605
606
607
608
609
610
611
612
613
614
def iter_users(self, **kwargs: dict) -> Generator[Dict[str, Any], None, None]:
    """ Iterate overs the platform users \n
        https://apiv1.lumapps.com/#operation/User/List

        Args:
            **kwargs: args to add to the request (see api documentation)

        Returns:
            A generator return the platform users
    """
    params = {"instance": self.instance_id}
    params.update(kwargs)
    yield from self.iter_call("user/list", **params)

save_user(self, user) ¤

Save a user

Parameters:

Name Type Description Default
user Dict[str, Any]

The user to save

required

Returns:

Type Description
Dict[str, Any]

The saved user

Source code in lumapps/api/client.py
621
622
623
624
625
626
627
628
629
630
631
632
633
def save_user(self, user: Dict[str, Any]) -> Dict[str, Any]:
    """ Save a user

        Args:
            user: The user to save

        Returns:
            The saved user
    """
    debug(f"Saving user: {to_json(user)}")
    if self.dry_run:
        return user
    return self.get_call("user/save", body=user)

Group¤

get_group(self, group_id) ¤

Get a group by his id

Parameters:

Name Type Description Default
group_id str

The id of the group

required

Returns:

Type Description
Dict[str, Any]

The retrieved group

Source code in lumapps/api/client.py
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
@lru_cache()
def get_group(self, group_id: str) -> Dict[str, Any]:
    """ Get a group by his id

        Args:
            group_id: The id of the group

        Returns:
            The retrieved group
    """
    return self.get_call("feed/get", uid=group_id)

iter_groups(self, type_id) ¤

Iter the groups on the current site

Parameters:

Name Type Description Default
type_id str

The id of the group type (feedType or category) to filter on

required

Returns:

Type Description
Generator[Dict[str, Any], NoneType, NoneType]

The groups in that group type

Source code in lumapps/api/client.py
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
def iter_groups(self, type_id: str) -> Generator[Dict[str, Any], None, None]:
    """ Iter the groups on the current site

        Args:
            type_id: The id of the group type (feedType or category) to filter on

        Returns:
            The groups in that group type
    """
    return self.iter_call(
        "feed/list", instance=self.instance_id, type=type_id, action="GROUP_EDIT"
    )

save_group(self, group, retries=0) ¤

Save a group

Parameters:

Name Type Description Default
group Dict[str, Any]

The group to save

required
retries int

The number of retries on failure to do

0

Returns:

Type Description
Dict[str, Any]

The saved group

Source code in lumapps/api/client.py
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
def save_group(self, group: Dict[str, Any], retries: int = 0) -> Dict[str, Any]:
    """ Save a group

        Args:
            group: The group to save
            retries: The number of retries on failure to do

        Returns:
            The saved group
    """
    info(f"Saving group: {to_json(group)}")
    if self.dry_run:
        return group
    attempt = 0
    while True:
        attempt += 1
        try:
            return self.get_call("feed/save", body=group)
        except HTTPStatusError as e:
            if e.response.status_code == 503 and attempt < 1 + retries:
                sleep_time = (attempt + 1) * 3
                warning(f"503 saving the feed, will retry in {sleep_time}s")
                sleep(sleep_time)
                continue
            raise