in src/dfcx_scrapi/core/scrapi_base.py [0:0]
def _parse_resource_path(
resource_type,
resource_id,
validate=True) -> Dict[str, str]:
# pylint: disable=line-too-long
"""Validates the provided Resource ID against known patterns.
Args:
resource_type, Must be one of the following resource types:
`agent`, `entity_type`, `environmnet`, `flow`, `intent`, `page`,
`project`, `security_setting`, `session`, `session_entity_type`,
`test_case`, `transition_route_group`, `version`, `webhook`
resource_id, The CX resource ID to check against the provided
resource_type
validate, allows the user to have their Resource ID validated along
with returning the parts dictionary of the Resource ID. If set to
True, this method will prompt the user with the correct format
to utilize for the specified ID. If set to False, no validation
will occur. If the input Resource ID is invalid when set to False,
an empty Dictionary will be returned, allowing the caller to
define their own ValueError message in a higher level class.
Defaults to True.
"""
data_store_match = r"[\w-]{1,947}"
engine_match = r"[a-z0-9][a-z0-9-_]{0,62}"
entity_id_match = r"[-@.0-9a-z]{1,36}"
location_id_match = r"[-0-9a-z]{1,36}"
page_id_match = r"[-0-9a-f]{1,36}|START_PAGE|END_SESSION|END_FLOW"
session_id_match = r"[-0-9a-zA-Z!@#$%^&*()_+={}[\]:;\"'<>,.?]{1,36}"
standard_id_match = r"[-0-9a-f]{1,36}"
version_id_match = r"[0-9]{1,4}"
matcher_root = f"^projects/(?P<project>.+?)/locations/(?P<location>{location_id_match})" # noqa: E501
pattern_map = {
"agent": {
"matcher": fr"{matcher_root}/agents/(?P<agent>{standard_id_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/agents/<Agent ID>`", # noqa: E501
},
"data_store": {
"matcher": fr"{matcher_root}/collections/default_collection/dataStores/(?P<data_store>{data_store_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/collections/default_collection/dataStores/<Data Store ID>`" # noqa: E501
},
"engine": {
"matcher": fr"{matcher_root}/collections/default_collection/engines/(?P<engine>{engine_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/collections/default_collection/engines/<Engine ID>`" # noqa: E501
},
"entity_type": {
"matcher": fr"{matcher_root}/agents/(?P<agent>{standard_id_match})/entityTypes/(?P<entity>{entity_id_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/agents/<Agent ID>/entityTypes/<Entity Types ID>`", # noqa: E501
},
"environment": {
"matcher": fr"{matcher_root}/agents/(?P<agent>{standard_id_match})/environments/(?P<environment>{standard_id_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/agents/<Agent ID>/environments/<Environment ID>`", # noqa: E501
},
"flow": {
"matcher": fr"{matcher_root}/agents/(?P<agent>{standard_id_match})/flows/(?P<flow>{standard_id_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/agents/<Agent ID>/flows/<Flow ID>`", # noqa: E501
},
"intent": {
"matcher": fr"{matcher_root}/agents/(?P<agent>{standard_id_match})/intents/(?P<intent>{standard_id_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/agents/<Agent ID>/intents/<Intent ID>`", # noqa: E501
},
"page": {
"matcher": fr"{matcher_root}/agents/(?P<agent>{standard_id_match})/flows/(?P<flow>{standard_id_match})/pages/(?P<page>{page_id_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/agents/<Agent ID>/flows/<Flow ID>/pages/<Page ID>`", # noqa: E501
},
"project": {
"matcher": fr"{matcher_root}$",
"format": "`projects/<Project ID>/locations/<Location ID>/`",
},
"security_setting": {
"matcher": fr"{matcher_root}/securitySettings/(?P<security_setting>{standard_id_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/securitySettings/<Security Setting ID>`", # noqa: E501
},
"session": {
"matcher": fr"{matcher_root}/agents/(?P<agent>{standard_id_match})(?:/environments/(?P<environment>{standard_id_match}))?/sessions/(?P<session>{session_id_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/agents/<Agent ID>/sessions/<Session ID>` or `projects/<Project ID>/locations/<Location ID>/agents/<Agent ID>/environments/<Environment ID>/sessions/<Session ID>`", # noqa: E501
},
"session_entity_type": {
"matcher": fr"{matcher_root}/agents/(?P<agent>{standard_id_match})/sessions/(?P<session>{session_id_match})/entityTypes/(?P<entity>{entity_id_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/agents/<Agent ID>/sessions/<Session ID>/entityTypes/<Entity Type ID>`", # noqa: E501
},
"test_case": {
"matcher": fr"{matcher_root}/agents/(?P<agent>{standard_id_match})/testCases/(?P<test_case>{standard_id_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/agents/<Agent ID>/testCases/<Test Case ID>`", # noqa: E501
},
"transition_route_group": {
"matcher": fr"{matcher_root}/agents/(?P<agent>{standard_id_match})/flows/(?P<flow>{standard_id_match})/transitionRouteGroups/(?P<transition_route_group>{standard_id_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/agents/<Agent ID>/flows/<Flow ID>/transitionRouteGroups/<Transition Route Group ID>`", # noqa: E501
},
"version": {
"matcher": fr"{matcher_root}/agents/(?P<agent>{standard_id_match})/flows/(?P<flow>{standard_id_match})/versions/(?P<version>{version_id_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/agents/<Agent ID>/flows/<Flow ID>`", # noqa: E501
},
"webhook": {
"matcher": fr"{matcher_root}/agents/(?P<agent>{standard_id_match})/webhooks/(?P<webhook>{standard_id_match})$", # noqa: E501
"format": "`projects/<Project ID>/locations/<Location ID>/agents/<Agent ID>`", # noqa: E501
}
}
if resource_type not in pattern_map:
raise KeyError(
"`resource_type` must be one of the following resource types:"
" `agent`, `data_store`, `engine`, `entity_type`, "
"`environmnet`, `flow`, `intent`, `page`, `project`, "
"`security_setting`, `session`, `session_entity_type`, "
"`test_case`, `transition_route_group`, `version`, `webhook`"
)
match_res = re.match(pattern_map[resource_type]["matcher"], resource_id)
dict_res = match_res.groupdict() if match_res else {}
valid_parse = False
if dict_res:
valid_parse = True
if validate and not valid_parse:
raise ValueError(
f"{resource_type.capitalize()} ID must be provided in the "
f"following format: "
f"{pattern_map[resource_type]['format']}"
)
# pylint: enable=line-too-long
return dict_res