Source code for idem_vra.states.vra.pipeline.customintegrations

from idem_vra.helpers.mapper import add_properties
from idem_vra.helpers.mapper import omit_properties
from idem_vra.helpers.models import StateReturn


__contracts__ = ["resource"]

TREQ = {"present": {"require": ["vra.iaas.project.present"]}, "absent": {"require": []}}


[docs]async def present(hub, ctx, name: str, **kwargs): """ :param string name: (required in body) A human-friendly name used as an identifier in APIs that support this option :param string apiVersion: (optional in query) The version of the API in yyyy-MM-dd format (UTC). For versioning information please refer to /codestream/api/about :param string Authorization: (optional in header) Bearer token :param string description: (optional in body) A human-friendly description. :param string project: (optional in body) The project this entity belongs to. :param string yaml: (optional in body) YAML describing Custom integration details. """ try: state = CustomintegrationsStateImpl(hub, ctx) return await state.present(hub, ctx, name, **kwargs) except Exception as error: hub.log.error("Error during enforcing present state: customintegrations") hub.log.error(str(error)) raise error
[docs]async def absent(hub, ctx, name: str, **kwargs): """ :param string p_id: (required in path) The ID of the Custom Integration :param string p_version: (required in path) The version of the Custom Integration :param string apiVersion: (optional in query) The version of the API in yyyy-MM-dd format (UTC). For versioning information please refer to /codestream/api/about :param string Authorization: (optional in header) Bearer token """ """ :param string name: (required) name of the resource """ try: state = CustomintegrationsStateImpl(hub, ctx) return await state.absent(hub, ctx, name, **kwargs) except Exception as error: hub.log.error("Error during enforcing absent state: customintegrations") hub.log.error(str(error)) raise error
[docs]async def describe(hub, ctx): try: state = CustomintegrationsStateImpl(hub, ctx) return await state.describe(hub, ctx) except Exception as error: hub.log.error("Error during describe: customintegrations") hub.log.error(str(error)) raise error
[docs]def is_pending(hub, ret: dict, state: str = None, **pending_kwargs): try: state = CustomintegrationsStateImpl(hub, None) return state.is_pending(hub, ret, state, **pending_kwargs) except Exception as error: hub.log.error("Error during is_pending: customintegrations") hub.log.error(str(error)) raise error
[docs]class CustomintegrationsState: def __init__(self, hub, ctx): self.hub = hub self.ctx = ctx
[docs] async def present(self, hub, ctx, name: str, **kwargs): search_result = (await self.paginate_find(hub, ctx))["ret"] for s in search_result.content: if name == s["name"] and True: hub.log.info( f'Returning resource customintegrations "{s["name"]}" due to existing resource "{name}"' ) s = await self.remap_resource_structure(hub, ctx, s) return StateReturn( result=True, comment=f"Resource customintegrations {name} already exists.", old=s, new=s, ) res = ( await hub.exec.vra.pipeline.customintegrations.create_custom_integration_using_post( ctx, name, **kwargs ) )["ret"] res = await self.remap_resource_structure(hub, ctx, res) return StateReturn( result=True, comment=f"Creation of customintegrations {name} success.", old=None, new=res, )
[docs] async def absent(self, hub, ctx, name: str, **kwargs): search_result = (await self.paginate_find(hub, ctx))["ret"] resource = None for s in search_result.content: if name == s["name"] and True: hub.log.info( f'Found resource customintegrations "{s["name"]}" due to existing resource "{name}"' ) s = await self.remap_resource_structure(hub, ctx, s) resource = s if resource: # it exists! delete_kwargs = {} delete_kwargs["p_id"] = resource.get("id") delete_kwargs["p_version"] = resource.get("version") hub.log.debug( f"customintegrations with name = {resource.get('name')} already exists" ) await hub.exec.vra.pipeline.customintegrations.delete_version_by_id_using_delete( ctx, **delete_kwargs ) return StateReturn( result=True, comment=f"Resource with name = {resource.get('name')} deleted.", old=resource, new=None, ) return StateReturn( result=True, comment=f"Resource with name = {name} is already absent.", old=None, new=None, )
[docs] async def describe(self, hub, ctx): result = {} res = await self.paginate_find(hub, ctx) for obj in res.get("ret", {}).get("content", []): # Keep track of name and id properties as they may get remapped obj_name = obj.get("name", "unknown") obj_id = obj.get("id", "unknown") obj = await self.remap_resource_structure(hub, ctx, obj) # Define props props = [{key: value} for key, value in obj.items()] # Build result result[f"{obj_name}-{obj_id.split('-')[-1]}"] = { "vra.pipeline.customintegrations.present": props } return result
[docs] async def paginate_find(self, hub, ctx, **kwargs): """ Paginate through all resources using their 'find' method. """ res = await hub.exec.vra.pipeline.customintegrations.get_custom_integrations_using_get( ctx, **kwargs ) numberOfElements = res.get("ret", {}).get("numberOfElements", 0) totalElements = res.get("ret", {}).get("totalElements", 0) initialElements = numberOfElements if numberOfElements != totalElements and totalElements != 0: while initialElements < totalElements: hub.log.debug( f"Requesting customintegrations with offset={initialElements} out of {totalElements}" ) pres = await hub.exec.vra.pipeline.customintegrations.get_custom_integrations_using_get( ctx, skip=initialElements ) initialElements += pres.get("ret", {}).get("numberOfElements", 0) aggO = res.get("ret", {}).get("content", []) aggN = pres.get("ret", {}).get("content", []) res["ret"]["content"] = [*aggO, *aggN] res["ret"]["numberOfElements"] = initialElements return res
[docs] def is_pending(self, hub, ret: dict, state: str = None, **pending_kwargs): """ State reconciliation """ hub.log.debug(f'Running is_pending for resource: {ret.get("__id__", None)}...') is_pending_result = False hub.log.debug( f'is_pending_result for resource "{ret.get("__id__", None)}": {is_pending_result}' ) return is_pending_result
[docs] async def remap_resource_structure(self, hub, ctx, obj: dict) -> dict: schema_mapper = { "add": [], "omit": [ "createdAt", "_createTimeInMicros", "_updateTimeInMicros", "updatedAt", "updatedBy", "createdBy", "_link", "_projectId", "version", "status", "sourceVersion", "changeLog", "parentId", "tags", ], } # Perform resource mapping by adding properties and omitting properties. # Property renaming is addition followed by omission. if schema_mapper: resource_name = "customintegrations" hub.log.debug(f"Remapping resource {resource_name}...") obj = await add_properties(obj, schema_mapper.get("add", [])) obj = omit_properties(obj, schema_mapper.get("omit", [])) return obj
# ==================================== # State override # ====================================
[docs]class CustomintegrationsStateImpl(CustomintegrationsState):
[docs] async def present(self, hub, ctx, name: str, **kwargs): search_result = (await self.paginate_find(hub, ctx))["ret"] for itm in search_result.get("links", []): s = search_result.documents.get(itm) if name == s.get("name", "") and True: hub.log.info( f'Returning resource customintegrations "{s["name"]}" due to existing resource "{name}"' ) s = await self.remap_resource_structure(hub, ctx, s) return StateReturn( result=True, comment=f"Resource customintegrations {name} already exists.", old=s, new=s, ) res = ( await hub.exec.vra.pipeline.customintegrations.create_custom_integration_using_post( ctx, name, **kwargs ) )["ret"] res = await self.remap_resource_structure(hub, ctx, res) return StateReturn( result=True, comment=f"Creation of customintegrations {name} success.", old=None, new=res, )
[docs] async def absent(self, hub, ctx, name: str, **kwargs): search_result = (await self.paginate_find(hub, ctx))["ret"] resource = None for itm in search_result.get("links", []): s = search_result.documents.get(itm) if name == s.get("name", "") and True: hub.log.info( f'Found resource customintegrations "{s["name"]}" due to existing resource "{name}"' ) s = await self.remap_resource_structure(hub, ctx, s) resource = s if resource: # it exists! delete_kwargs = {} delete_kwargs["p_id"] = resource.get("id") delete_kwargs["p_version"] = resource.get("version") hub.log.debug( f"customintegrations with name = {resource.get('name')} already exists" ) await hub.exec.vra.pipeline.customintegrations.delete_version_by_id_using_delete( ctx, **delete_kwargs ) return StateReturn( result=True, comment=f"Resource with name = {resource.get('name')} deleted.", old=resource, new=None, ) return StateReturn( result=True, comment=f"Resource with name = {name} is already absent.", old=None, new=None, )
[docs] async def describe(self, hub, ctx): result = {} res = await self.paginate_find(hub, ctx) for itm in res.get("ret", {}).get("links", []): # Keep track of name and id properties as they may get remapped obj = res.get("ret", {}).get("documents", {}).get(itm) obj_name = obj.get("name", "unknown") obj_id = obj.get("id", "unknown") obj = await self.remap_resource_structure(hub, ctx, obj) # Define props props = [{key: value} for key, value in obj.items()] # Build result result[f"{obj_name}-{obj_id.split('-')[-1]}"] = { "vra.pipeline.customintegrations.present": props } return result