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 |
|
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 182def 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 216def 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 393def 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 614def 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 633def 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 1627def 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 1684def 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