diff --git a/plugins/aws/fix_plugin_aws/collector.py b/plugins/aws/fix_plugin_aws/collector.py index 16e6ab1db9..4256c5d30d 100644 --- a/plugins/aws/fix_plugin_aws/collector.py +++ b/plugins/aws/fix_plugin_aws/collector.py @@ -9,6 +9,7 @@ from fix_plugin_aws.aws_client import AwsClient from fix_plugin_aws.configuration import AwsConfig from fix_plugin_aws.resource import ( + amazonq, apigateway, athena, autoscaling, @@ -105,6 +106,7 @@ + sqs.resources + redshift.resources + backup.resources + + amazonq.resources ) all_resources: List[Type[AwsResource]] = global_resources + regional_resources diff --git a/plugins/aws/fix_plugin_aws/resource/amazonq.py b/plugins/aws/fix_plugin_aws/resource/amazonq.py new file mode 100644 index 0000000000..19e71a1fc4 --- /dev/null +++ b/plugins/aws/fix_plugin_aws/resource/amazonq.py @@ -0,0 +1,1088 @@ +from datetime import datetime +from typing import ClassVar, Dict, Optional, Type, List, Any + +from attrs import define, field + +from fix_plugin_aws.resource.base import AwsResource, GraphBuilder, AwsApiSpec +from fix_plugin_aws.aws_client import AwsClient +from fix_plugin_aws.utils import TagsValue, ToDict +from fixlib.baseresources import ModelReference +from fixlib.graph import Graph +from fixlib.json_bender import Bender, S, Bend, ForallBend +from fixlib.types import Json + +service_name = "qbusiness" + + +class AmazonQTaggable: + def update_resource_tag(self, client: AwsClient, key: str, value: str) -> bool: + if isinstance(self, AwsResource): + client.call( + aws_service=service_name, + action="tag-resource", + result_name=None, + resourceARN=self.arn, + tags=[{"key": key, "value": value}], + ) + return True + return False + + def delete_resource_tag(self, client: AwsClient, key: str) -> bool: + if isinstance(self, AwsResource): + client.call( + aws_service=service_name, + action="untag-resource", + result_name=None, + resourceARN=self.arn, + tagKeys=[key], + ) + return True + return False + + @classmethod + def service_name(cls) -> str: + return service_name + + +@define(eq=False, slots=False) +class AwsQBusinessApplication(AmazonQTaggable, AwsResource): + kind: ClassVar[str] = "aws_q_business_application" + kind_display: ClassVar[str] = "AWS QBusiness Application" + kind_description: ClassVar[str] = ( + "Represents a QBusiness application within the AWS QBusiness service. Applications" + " define a set of tasks and configuration for processing data within the QBusiness ecosystem." + ) + aws_metadata: ClassVar[Dict[str, Any]] = { + "provider_link_tpl": "https://{region_id}.console.aws.amazon.com/amazonq/business/applications/{id}/details?region={region}", # fmt: skip + "arn_tpl": "arn:{partition}:qbusiness:{region}:{account}:application/{id}", + } + reference_kinds: ClassVar[ModelReference] = { + "successors": { + "default": [ + "aws_q_business_conversation", + "aws_q_business_data_source", + "aws_q_business_data_source_sync_job", + "aws_q_business_document", + "aws_q_business_indice", + "aws_q_business_message", + "aws_q_business_plugin", + "aws_q_business_retriever", + "aws_q_business_web_experience", + "aws_q_apps_library_item", + "aws_q_apps", + ] + }, + } + api_spec: ClassVar[AwsApiSpec] = AwsApiSpec("qbusiness", "list-applications", "applications") + mapping: ClassVar[Dict[str, Bender]] = { + "id": S("applicationId"), + "name": S("displayName"), + "ctime": S("createdAt"), + "mtime": S("updatedAt"), + "display_name": S("displayName"), + "application_id": S("applicationId"), + "created_at": S("createdAt"), + "updated_at": S("updatedAt"), + "status": S("status"), + } + display_name: Optional[str] = field(default=None, metadata={"description": "The name of the Amazon Q Business application."}) # fmt: skip + application_id: Optional[str] = field(default=None, metadata={"description": "The identifier for the Amazon Q Business application."}) # fmt: skip + created_at: Optional[datetime] = field(default=None, metadata={"description": "The Unix timestamp when the Amazon Q Business application was created."}) # fmt: skip + updated_at: Optional[datetime] = field(default=None, metadata={"description": "The Unix timestamp when the Amazon Q Business application was last updated."}) # fmt: skip + status: Optional[str] = field(default=None, metadata={"description": "The status of the Amazon Q Business application. The application is ready to use when the status is ACTIVE."}) # fmt: skip + + @classmethod + def called_collect_apis(cls) -> List[AwsApiSpec]: + return [ + cls.api_spec, + AwsApiSpec(service_name, "list-tags-for-resource"), + ] + + @classmethod + def called_mutator_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "tag-resource"), + AwsApiSpec(service_name, "untag-resource"), + AwsApiSpec(service_name, "delete-application"), + ] + + def delete_resource(self, client: AwsClient, graph: Graph) -> bool: + client.call( + aws_service="qbusiness", + action="delete-application", + result_name=None, + applicationId=self.id, + ) + return True + + @classmethod + def collect(cls: Type[AwsResource], json: List[Json], builder: GraphBuilder) -> None: + def add_tags(tag_resource: AwsResource) -> None: + # Filter resources that have tags + if not isinstance( + tag_resource, + ( + AwsQBusinessApplication, + AwsQBusinessDataSource, + AwsQBusinessIndice, + AwsQBusinessPlugin, + AwsQBusinessRetriever, + AwsQBusinessWebExperience, + AwsQApps, + ), + ): + return + if isinstance(tag_resource, AwsQApps): + tags = builder.client.list( + "qapps", + "list-tags-for-resource", + "tags", + expected_errors=["ResourceNotFoundException"], + resourceARN=tag_resource.arn, + ) + if tags: + tag_resource.tags.update(tags) + else: + tags = builder.client.list( + service_name, + "list-tags-for-resource", + "tags", + expected_errors=["ResourceNotFoundException"], + resourceARN=tag_resource.arn, + ) + if tags: + for tag in tags: + tag_resource.tags.update({tag.get("key"): tag.get("value")}) + + def collect_application_resources( + application: AwsQBusinessApplication, + resource_class: AwsResource, + action: str, + result_name: str, + param_name: str = "applicationId", + service: Optional[str] = None, + ) -> None: + param_map = {param_name: application.id} + q_resources = builder.client.list( + service or service_name, + action, + result_name, + expected_errors=[], + **param_map, + ) + for q_resource in q_resources: + if resource := resource_class.from_api(q_resource, builder): + if isinstance(resource, (AwsQBusinessPlugin, AwsQBusinessWebExperience, AwsQBusinessIndice)): + resource.application_id = application.id + if isinstance(resource, (AwsQApps, AwsQAppsLibraryItem)): + resource.instance_id = application.id + builder.add_node(resource, q_resource) + builder.add_edge(application, node=resource) + builder.submit_work(service_name, add_tags, resource) + if isinstance(resource, AwsQBusinessConversation): + m_resources = builder.client.list( + service_name, + "list-messages", + "messages", + applicationId=application.id, + conversationId=resource.id, + ) + for message in m_resources: + if message_resource := AwsQBusinessMessage.from_api(message, builder): + builder.add_node(message_resource, message) + builder.add_edge(resource, node=message_resource) + + def collect_indice_resources(application: AwsQBusinessApplication) -> None: + def collect_index_resources( + indice: "AwsQBusinessIndice", resource_class: AwsResource, action: str, result_name: str + ) -> None: + i_resources = builder.client.list( + service_name, + action, + result_name, + applicationId=application.id, + indexId=indice.id, + ) + for i_resource in i_resources: + if resource := resource_class.from_api(i_resource, builder): + if isinstance(resource, AwsQBusinessDataSource): + resource.application_id = application.id + resource.indice_id = indice.id + builder.add_node(resource, i_resource) + builder.add_edge(indice, node=resource) + builder.submit_work(service_name, add_tags, resource) + if isinstance(resource, AwsQBusinessDataSource): + data_source_job_resources = builder.client.list( + service_name, + "list-data-source-sync-jobs", + "history", + applicationId=application.id, + indexId=indice.id, + dataSourceId=resource.id, + ) + for data_source_job in data_source_job_resources: + if sync_job := AwsQBusinessDataSourceSyncJob.from_api(data_source_job, builder): + builder.add_node(sync_job, data_source_job) + builder.add_edge(resource, node=sync_job) + + index_resources = builder.client.list( + service_name, + "list-indices", + "indices", + applicationId=application.id, + ) + for index_resource in index_resources: + if indice_instance := AwsQBusinessIndice.from_api(index_resource, builder): + indice_instance.application_id = application.id + builder.add_node(indice_instance, index_resource) + builder.add_edge(application, node=indice_instance) + builder.submit_work(service_name, add_tags, indice_instance) + builder.submit_work( + service_name, + collect_index_resources, + indice_instance, + AwsQBusinessDataSource, + "list-data-sources", + "dataSources", + ) + builder.submit_work( + service_name, + collect_index_resources, + indice_instance, + AwsQBusinessDocument, + "list-documents", + "documentDetailList", + ) + + for js in json: + if instance := cls.from_api(js, builder): + builder.add_node(instance, js) + builder.submit_work(service_name, add_tags, instance) + builder.submit_work( + service_name, + collect_application_resources, + instance, + AwsQBusinessConversation, + "list-conversations", + "conversations", + ) + builder.submit_work(service_name, collect_indice_resources, instance) + builder.submit_work( + service_name, collect_application_resources, instance, AwsQBusinessPlugin, "list-plugins", "plugins" + ) + builder.submit_work( + service_name, + collect_application_resources, + instance, + AwsQBusinessRetriever, + "list-retrievers", + "retrievers", + ) + builder.submit_work( + service_name, + collect_application_resources, + instance, + AwsQBusinessWebExperience, + "list-web-experiences", + "webExperiences", + ) + builder.submit_work( + service_name, + collect_application_resources, + instance, + AwsQAppsLibraryItem, + "list-library-items", + "libraryItems", + "instanceId", + "qapps", + ) + builder.submit_work( + service_name, + collect_application_resources, + instance, + AwsQApps, + "list-q-apps", + "apps", + "instanceId", + "qapps", + ) + + +@define(eq=False, slots=False) +class AwsQBusinessConversation(AwsResource): + kind: ClassVar[str] = "aws_q_business_conversation" + kind_display: ClassVar[str] = "AWS QBusiness Conversation" + kind_description: ClassVar[str] = ( + "Represents a conversation within the AWS QBusiness service. Conversations are" + " interactions that involve a series of messages or data exchanges." + ) + # Collected via AwsQBusinessApplication() + mapping: ClassVar[Dict[str, Bender]] = { + "id": S("conversationId"), + "tags": S("Tags", default=[]) >> ToDict(), + "name": S("Tags", default=[]) >> TagsValue("Name"), + "conversation_id": S("conversationId"), + "title": S("title"), + "start_time": S("startTime"), + } + conversation_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the Amazon Q Business conversation."}) # fmt: skip + title: Optional[str] = field(default=None, metadata={"description": "The title of the conversation."}) # fmt: skip + start_time: Optional[datetime] = field(default=None, metadata={"description": "The start time of the conversation."}) # fmt: skip + + @classmethod + def called_collect_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "list-conversations", "conversations"), + ] + + @classmethod + def service_name(cls) -> str: + return service_name + + +@define(eq=False, slots=False) +class AwsQBusinessDataSource(AmazonQTaggable, AwsResource): + kind: ClassVar[str] = "aws_q_business_data_source" + kind_display: ClassVar[str] = "AWS QBusiness Data Source" + kind_description: ClassVar[str] = ( + "Represents a data source in the AWS QBusiness service. Data sources are the origins" + " from which data is ingested for processing or analysis within the QBusiness framework." + ) + # Collected via AwsQBusinessApplication() + aws_metadata: ClassVar[Dict[str, Any]] = { + "provider_link_tpl": "https://{region_id}.console.aws.amazon.com/amazonq/business/applications/{app_id}/indices/{indice_id}/datasources/{id}/details?region={region}", # fmt: skip + "arn_tpl": "arn:{partition}:qbusiness:{region}:{account}:application/{application_id}/index/{indice_id}/data-source/{id}", + "extra_args": ["application_id", "indice_id"], + } + mapping: ClassVar[Dict[str, Bender]] = { + "id": S("dataSourceId"), + "tags": S("Tags", default=[]) >> ToDict(), + "name": S("displayName"), + "ctime": S("createdAt"), + "mtime": S("updatedAt"), + "display_name": S("displayName"), + "data_source_id": S("dataSourceId"), + "type": S("type"), + "created_at": S("createdAt"), + "updated_at": S("updatedAt"), + "status": S("status"), + } + indice_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the Amazon Q Business indice."}) # fmt: skip + application_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the Amazon Q Business application."}) # fmt: skip + display_name: Optional[str] = field(default=None, metadata={"description": "The name of the Amazon Q Business data source."}) # fmt: skip + data_source_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the Amazon Q Business data source."}) # fmt: skip + type: Optional[str] = field(default=None, metadata={"description": "The type of the Amazon Q Business data source."}) # fmt: skip + created_at: Optional[datetime] = field(default=None, metadata={"description": "The Unix timestamp when the Amazon Q Business data source was created."}) # fmt: skip + updated_at: Optional[datetime] = field(default=None, metadata={"description": "The Unix timestamp when the Amazon Q Business data source was last updated."}) # fmt: skip + status: Optional[str] = field(default=None, metadata={"description": "The status of the Amazon Q Business data source."}) # fmt: skip + + @classmethod + def called_collect_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "list-data-sources", "dataSources"), + AwsApiSpec(service_name, "list-tags-for-resource"), + ] + + @classmethod + def called_mutator_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "tag-resource"), + AwsApiSpec(service_name, "untag-resource"), + AwsApiSpec(service_name, "delete-data-source"), + ] + + def delete_resource(self, client: AwsClient, graph: Graph) -> bool: + client.call( + aws_service="qbusiness", + action="delete-data-source", + result_name=None, + applicationId=self.application_id, + indexId=self.indice_id, + dataSourceId=self.id, + ) + return True + + +@define(eq=False, slots=False) +class AwsQBusinessErrorDetail: + kind: ClassVar[str] = "aws_q_business_error_detail" + mapping: ClassVar[Dict[str, Bender]] = {"error_message": S("errorMessage"), "error_code": S("errorCode")} + error_message: Optional[str] = field(default=None, metadata={"description": "The message explaining the data source sync error."}) # fmt: skip + error_code: Optional[str] = field(default=None, metadata={"description": "The code associated with the data source sync error."}) # fmt: skip + + +@define(eq=False, slots=False) +class AwsQBusinessDataSourceSyncJobMetrics: + kind: ClassVar[str] = "aws_q_business_data_source_sync_job_metrics" + mapping: ClassVar[Dict[str, Bender]] = { + "documents_added": S("documentsAdded"), + "documents_modified": S("documentsModified"), + "documents_deleted": S("documentsDeleted"), + "documents_failed": S("documentsFailed"), + "documents_scanned": S("documentsScanned"), + } + documents_added: Optional[str] = field(default=None, metadata={"description": "The current count of documents added from the data source during the data source sync."}) # fmt: skip + documents_modified: Optional[str] = field(default=None, metadata={"description": "The current count of documents modified in the data source during the data source sync."}) # fmt: skip + documents_deleted: Optional[str] = field(default=None, metadata={"description": "The current count of documents deleted from the data source during the data source sync."}) # fmt: skip + documents_failed: Optional[str] = field(default=None, metadata={"description": "The current count of documents that failed to sync from the data source during the data source sync."}) # fmt: skip + documents_scanned: Optional[str] = field(default=None, metadata={"description": "The current count of documents crawled by the ongoing sync job in the data source."}) # fmt: skip + + +@define(eq=False, slots=False) +class AwsQBusinessDataSourceSyncJob(AwsResource): + kind: ClassVar[str] = "aws_q_business_data_source_sync_job" + kind_display: ClassVar[str] = "AWS QBusiness Data Source Sync Job" + kind_description: ClassVar[str] = ( + "Represents a data source synchronization job in the AWS QBusiness service. Sync jobs" + " ensure that data from data sources is up-to-date and correctly integrated into the system." + ) + # Collected via AwsQBusinessApplication() + mapping: ClassVar[Dict[str, Bender]] = { + "id": S("executionId"), + "tags": S("Tags", default=[]) >> ToDict(), + "name": S("Tags", default=[]) >> TagsValue("Name"), + "execution_id": S("executionId"), + "start_time": S("startTime"), + "end_time": S("endTime"), + "status": S("status"), + "sync_job_error": S("error") >> Bend(AwsQBusinessErrorDetail.mapping), + "data_source_error_code": S("dataSourceErrorCode"), + "sync_job_metrics": S("metrics") >> Bend(AwsQBusinessDataSourceSyncJobMetrics.mapping), + } + execution_id: Optional[str] = field(default=None, metadata={"description": "The identifier of a data source synchronization job."}) # fmt: skip + start_time: Optional[datetime] = field(default=None, metadata={"description": "The Unix time stamp when the data source synchronization job started."}) # fmt: skip + end_time: Optional[datetime] = field(default=None, metadata={"description": "The Unix timestamp when the synchronization job completed."}) # fmt: skip + status: Optional[str] = field(default=None, metadata={"description": "The status of the synchronization job. When the Status field is set to SUCCEEDED, the synchronization job is done. If the status code is FAILED, the ErrorCode and ErrorMessage fields give you the reason for the failure."}) # fmt: skip + sync_job_error: Optional[AwsQBusinessErrorDetail] = field(default=None, metadata={"description": "If the Status field is set to FAILED, the ErrorCode field indicates the reason the synchronization failed."}) # fmt: skip + data_source_error_code: Optional[str] = field(default=None, metadata={"description": "If the reason that the synchronization failed is due to an error with the underlying data source, this field contains a code that identifies the error."}) # fmt: skip + sync_job_metrics: Optional[AwsQBusinessDataSourceSyncJobMetrics] = field(default=None, metadata={"description": "Maps a batch delete document request to a specific data source sync job. This is optional and should only be supplied when documents are deleted by a data source connector."}) # fmt: skip + + @classmethod + def called_collect_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "list-data-source-sync-jobs", "history"), + ] + + @classmethod + def service_name(cls) -> str: + return service_name + + +@define(eq=False, slots=False) +class AwsQBusinessDocument(AwsResource): + kind: ClassVar[str] = "aws_q_business_document" + kind_display: ClassVar[str] = "AWS QBusiness Document" + kind_description: ClassVar[str] = ( + "Represents a document within the AWS QBusiness service. Documents are structured pieces" + " of information that can be used for various purposes within the QBusiness ecosystem." + ) + # Collected via AwsQBusinessApplication() + mapping: ClassVar[Dict[str, Bender]] = { + "id": S("documentId"), + "tags": S("Tags", default=[]) >> ToDict(), + "name": S("Tags", default=[]) >> TagsValue("Name"), + "ctime": S("createdAt"), + "mtime": S("updatedAt"), + "document_id": S("documentId"), + "status": S("status"), + "document_error": S("error") >> Bend(AwsQBusinessErrorDetail.mapping), + "created_at": S("createdAt"), + "updated_at": S("updatedAt"), + } + document_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the document."}) # fmt: skip + status: Optional[str] = field(default=None, metadata={"description": "The current status of the document."}) # fmt: skip + document_error: Optional[AwsQBusinessErrorDetail] = field(default=None, metadata={"description": "An error message associated with the document."}) # fmt: skip + created_at: Optional[datetime] = field(default=None, metadata={"description": "The timestamp for when the document was created."}) # fmt: skip + updated_at: Optional[datetime] = field(default=None, metadata={"description": "The timestamp for when the document was last updated."}) # fmt: skip + + @classmethod + def called_collect_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "list-documents", "documentDetailList"), + ] + + @classmethod + def service_name(cls) -> str: + return service_name + + +@define(eq=False, slots=False) +class AwsQBusinessIndice(AmazonQTaggable, AwsResource): + kind: ClassVar[str] = "aws_q_business_indice" + kind_display: ClassVar[str] = "AWS QBusiness Indice" + kind_description: ClassVar[str] = ( + "Represents an index in the AWS QBusiness service. Indices are used to organize and" + " facilitate efficient searching and retrieval of data within the QBusiness framework." + ) + aws_metadata: ClassVar[Dict[str, Any]] = { + "arn_tpl": "arn:{partition}:qbusiness:{region}:{account}:application/{application_id}/index/{id}", + "extra_args": ["application_id"], + } + # Collected via AwsQBusinessApplication() + mapping: ClassVar[Dict[str, Bender]] = { + "id": S("indexId"), + "tags": S("Tags", default=[]) >> ToDict(), + "name": S("displayName"), + "ctime": S("createdAt"), + "mtime": S("updatedAt"), + "display_name": S("displayName"), + "index_id": S("indexId"), + "created_at": S("createdAt"), + "updated_at": S("updatedAt"), + "status": S("status"), + } + application_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the Amazon Q Business application."}) # fmt: skip + display_name: Optional[str] = field(default=None, metadata={"description": "The name of the index."}) # fmt: skip + index_id: Optional[str] = field(default=None, metadata={"description": "The identifier for the index."}) # fmt: skip + created_at: Optional[datetime] = field(default=None, metadata={"description": "The Unix timestamp when the index was created."}) # fmt: skip + updated_at: Optional[datetime] = field(default=None, metadata={"description": "The Unix timestamp when the index was last updated."}) # fmt: skip + status: Optional[str] = field(default=None, metadata={"description": "The current status of the index. When the status is ACTIVE, the index is ready."}) # fmt: skip + + @classmethod + def called_collect_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "list-indices", "indices"), + AwsApiSpec(service_name, "list-tags-for-resource"), + ] + + @classmethod + def called_mutator_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "tag-resource"), + AwsApiSpec(service_name, "untag-resource"), + AwsApiSpec(service_name, "delete-index"), + ] + + def delete_resource(self, client: AwsClient, graph: Graph) -> bool: + client.call( + aws_service="qbusiness", + action="delete-index", + result_name=None, + applicationId=self.application_id, + indexId=self.id, + ) + return True + + +@define(eq=False, slots=False) +class AwsQBusinessAttachmentOutput: + kind: ClassVar[str] = "aws_q_business_attachment_output" + mapping: ClassVar[Dict[str, Bender]] = { + "name": S("name"), + "status": S("status"), + "error": S("error") >> Bend(AwsQBusinessErrorDetail.mapping), + } + name: Optional[str] = field(default=None, metadata={"description": "The name of a file uploaded during chat."}) # fmt: skip + status: Optional[str] = field(default=None, metadata={"description": "The status of a file uploaded during chat."}) # fmt: skip + error: Optional[AwsQBusinessErrorDetail] = field(default=None, metadata={"description": "An error associated with a file uploaded during chat."}) # fmt: skip + + +@define(eq=False, slots=False) +class AwsQBusinessTextSegment: + kind: ClassVar[str] = "aws_q_business_text_segment" + mapping: ClassVar[Dict[str, Bender]] = { + "begin_offset": S("beginOffset"), + "end_offset": S("endOffset"), + "snippet_excerpt": S("snippetExcerpt", "text"), + } + begin_offset: Optional[int] = field(default=None, metadata={"description": "The zero-based location in the response string where the source attribution starts."}) # fmt: skip + end_offset: Optional[int] = field(default=None, metadata={"description": "The zero-based location in the response string where the source attribution ends."}) # fmt: skip + snippet_excerpt: Optional[str] = field(default=None, metadata={"description": "The relevant text excerpt from a source that was used to generate a citation text segment in an Amazon Q Business chat response."}) # fmt: skip + + +@define(eq=False, slots=False) +class AwsQBusinessSourceAttribution: + kind: ClassVar[str] = "aws_q_business_source_attribution" + mapping: ClassVar[Dict[str, Bender]] = { + "title": S("title"), + "snippet": S("snippet"), + "url": S("url"), + "citation_number": S("citationNumber"), + "updated_at": S("updatedAt"), + "text_message_segments": S("textMessageSegments", default=[]) >> ForallBend(AwsQBusinessTextSegment.mapping), + } + title: Optional[str] = field(default=None, metadata={"description": "The title of the document which is the source for the Amazon Q Business generated response."}) # fmt: skip + snippet: Optional[str] = field(default=None, metadata={"description": "The content extract from the document on which the generated response is based."}) # fmt: skip + url: Optional[str] = field(default=None, metadata={"description": "The URL of the document which is the source for the Amazon Q Business generated response."}) # fmt: skip + citation_number: Optional[int] = field(default=None, metadata={"description": "The number attached to a citation in an Amazon Q Business generated response."}) # fmt: skip + updated_at: Optional[datetime] = field(default=None, metadata={"description": "The Unix timestamp when the Amazon Q Business application was last updated."}) # fmt: skip + text_message_segments: Optional[List[AwsQBusinessTextSegment]] = field(factory=list, metadata={"description": "A text extract from a source document that is used for source attribution."}) # fmt: skip + + +@define(eq=False, slots=False) +class AwsQBusinessActionPayloadFieldValue: + kind: ClassVar[str] = "aws_q_business_action_payload_field_value" + mapping: ClassVar[Dict[str, Bender]] = {} + + +@define(eq=False, slots=False) +class AwsQBusinessActionReviewPayloadFieldAllowedValue: + kind: ClassVar[str] = "aws_q_business_action_review_payload_field_allowed_value" + mapping: ClassVar[Dict[str, Bender]] = { + "value": S("value") >> Bend(AwsQBusinessActionPayloadFieldValue.mapping), + "display_value": S("displayValue") >> Bend(AwsQBusinessActionPayloadFieldValue.mapping), + } + value: Optional[AwsQBusinessActionPayloadFieldValue] = field(default=None, metadata={"description": "The field value."}) # fmt: skip + display_value: Optional[AwsQBusinessActionPayloadFieldValue] = field(default=None, metadata={"description": "The name of the field."}) # fmt: skip + + +@define(eq=False, slots=False) +class AwsQBusinessActionReviewPayloadField: + kind: ClassVar[str] = "aws_q_business_action_review_payload_field" + mapping: ClassVar[Dict[str, Bender]] = { + "display_name": S("displayName"), + "display_order": S("displayOrder"), + "display_description": S("displayDescription"), + "type": S("type"), + "value": S("value") >> Bend(AwsQBusinessActionPayloadFieldValue.mapping), + "allowed_values": S("allowedValues", default=[]) + >> ForallBend(AwsQBusinessActionReviewPayloadFieldAllowedValue.mapping), + "allowed_format": S("allowedFormat"), + "required": S("required"), + } + display_name: Optional[str] = field(default=None, metadata={"description": "The name of the field."}) # fmt: skip + display_order: Optional[int] = field(default=None, metadata={"description": "The display order of fields in a payload."}) # fmt: skip + display_description: Optional[str] = field(default=None, metadata={"description": "The field level description of each action review input field. This could be an explanation of the field. In the Amazon Q Business web experience, these descriptions could be used to display as tool tips to help users understand the field."}) # fmt: skip + type: Optional[str] = field(default=None, metadata={"description": "The type of field."}) # fmt: skip + value: Optional[AwsQBusinessActionPayloadFieldValue] = field(default=None, metadata={"description": "The field value."}) # fmt: skip + allowed_values: Optional[List[AwsQBusinessActionReviewPayloadFieldAllowedValue]] = field(factory=list, metadata={"description": "Information about the field values that an end user can use to provide to Amazon Q Business for Amazon Q Business to perform the requested plugin action."}) # fmt: skip + allowed_format: Optional[str] = field(default=None, metadata={"description": "The expected data format for the action review input field value. For example, in PTO request, from and to would be of datetime allowed format."}) # fmt: skip + required: Optional[bool] = field(default=None, metadata={"description": "Information about whether the field is required."}) # fmt: skip + + +@define(eq=False, slots=False) +class AwsQBusinessActionReview: + kind: ClassVar[str] = "aws_q_business_action_review" + mapping: ClassVar[Dict[str, Bender]] = { + "plugin_id": S("pluginId"), + "plugin_type": S("pluginType"), + "payload": S("payload"), + "payload_field_name_separator": S("payloadFieldNameSeparator"), + } + plugin_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the plugin associated with the action review."}) # fmt: skip + plugin_type: Optional[str] = field(default=None, metadata={"description": "The type of plugin."}) # fmt: skip + payload: Optional[Dict[str, AwsQBusinessActionReviewPayloadField]] = field(default=None, metadata={"description": "Field values that an end user needs to provide to Amazon Q Business for Amazon Q Business to perform the requested plugin action."}) # fmt: skip + payload_field_name_separator: Optional[str] = field(default=None, metadata={"description": "A string used to retain information about the hierarchical contexts within an action review payload."}) # fmt: skip + + +@define(eq=False, slots=False) +class AwsQBusinessActionExecutionPayloadField: + kind: ClassVar[str] = "aws_q_business_action_execution_payload_field" + mapping: ClassVar[Dict[str, Bender]] = {"value": S("value") >> Bend(AwsQBusinessActionPayloadFieldValue.mapping)} + value: Optional[AwsQBusinessActionPayloadFieldValue] = field(default=None, metadata={"description": "The content of a user input field in an plugin action execution payload."}) # fmt: skip + + +@define(eq=False, slots=False) +class AwsQBusinessActionExecution: + kind: ClassVar[str] = "aws_q_business_action_execution" + mapping: ClassVar[Dict[str, Bender]] = { + "plugin_id": S("pluginId"), + "payload": S("payload"), + "payload_field_name_separator": S("payloadFieldNameSeparator"), + } + plugin_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the plugin the action is attached to."}) # fmt: skip + payload: Optional[Dict[str, AwsQBusinessActionExecutionPayloadField]] = field(default=None, metadata={"description": "A mapping of field names to the field values in input that an end user provides to Amazon Q Business requests to perform their plugin action."}) # fmt: skip + payload_field_name_separator: Optional[str] = field(default=None, metadata={"description": "A string used to retain information about the hierarchical contexts within an action execution event payload."}) # fmt: skip + + +@define(eq=False, slots=False) +class AwsQBusinessMessage(AwsResource): + kind: ClassVar[str] = "aws_q_business_message" + kind_display: ClassVar[str] = "AWS QBusiness Message" + kind_description: ClassVar[str] = ( + "Represents a message within the AWS QBusiness service. Messages are used for communication" + " or data exchange between various components or users within the QBusiness ecosystem." + ) + # Collected via AwsQBusinessApplication() + mapping: ClassVar[Dict[str, Bender]] = { + "id": S("messageId"), + "tags": S("Tags", default=[]) >> ToDict(), + "name": S("Tags", default=[]) >> TagsValue("Name"), + "message_id": S("messageId"), + "body": S("body"), + "time": S("time"), + "type": S("type"), + "message_attachments": S("attachments", default=[]) >> ForallBend(AwsQBusinessAttachmentOutput.mapping), + "source_attribution": S("sourceAttribution", default=[]) >> ForallBend(AwsQBusinessSourceAttribution.mapping), + "action_review": S("actionReview") >> Bend(AwsQBusinessActionReview.mapping), + "action_execution": S("actionExecution") >> Bend(AwsQBusinessActionExecution.mapping), + } + message_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the Amazon Q Business web experience message."}) # fmt: skip + body: Optional[str] = field(default=None, metadata={"description": "The content of the Amazon Q Business web experience message."}) # fmt: skip + time: Optional[datetime] = field(default=None, metadata={"description": "The timestamp of the first Amazon Q Business web experience message."}) # fmt: skip + type: Optional[str] = field(default=None, metadata={"description": "The type of Amazon Q Business message, whether HUMAN or AI generated."}) # fmt: skip + message_attachments: Optional[List[AwsQBusinessAttachmentOutput]] = field(factory=list, metadata={"description": "A file directly uploaded into an Amazon Q Business web experience chat."}) # fmt: skip + source_attribution: Optional[List[AwsQBusinessSourceAttribution]] = field(factory=list, metadata={"description": "The source documents used to generate Amazon Q Business web experience message."}) # fmt: skip + action_review: Optional[AwsQBusinessActionReview] = field(default=None, metadata={"description": "An output event that Amazon Q Business returns to an user who wants to perform a plugin action during a non-streaming chat conversation. It contains information about the selected action with a list of possible user input fields, some pre-populated by Amazon Q Business."}) # fmt: skip + action_execution: Optional[AwsQBusinessActionExecution] = field(default=None, metadata={"description": "Performs an Amazon Q Business plugin action during a non-streaming chat conversation."}) # fmt: skip + + @classmethod + def called_collect_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "list-messages", "messages"), + ] + + @classmethod + def service_name(cls) -> str: + return service_name + + +@define(eq=False, slots=False) +class AwsQBusinessPlugin(AmazonQTaggable, AwsResource): + kind: ClassVar[str] = "aws_q_business_plugin" + kind_display: ClassVar[str] = "AWS QBusiness Plugin" + kind_description: ClassVar[str] = ( + "Represents a plugin in the AWS QBusiness service. Plugins extend the functionality of" + " the QBusiness framework by adding new features or capabilities." + ) + aws_metadata: ClassVar[Dict[str, Any]] = { + "arn_tpl": "arn:{partition}:qbusiness:{region}:{account}:application/{application_id}/plugin/{id}", + "extra_args": ["application_id"], + } + # Collected via AwsQBusinessApplication() + mapping: ClassVar[Dict[str, Bender]] = { + "id": S("pluginId"), + "tags": S("Tags", default=[]) >> ToDict(), + "name": S("displayName"), + "ctime": S("createdAt"), + "mtime": S("updatedAt"), + "plugin_id": S("pluginId"), + "display_name": S("displayName"), + "type": S("type"), + "server_url": S("serverUrl"), + "state": S("state"), + "build_status": S("buildStatus"), + "created_at": S("createdAt"), + "updated_at": S("updatedAt"), + } + application_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the Amazon Q Business application."}) # fmt: skip + plugin_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the plugin."}) # fmt: skip + display_name: Optional[str] = field(default=None, metadata={"description": "The name of the plugin."}) # fmt: skip + type: Optional[str] = field(default=None, metadata={"description": "The type of the plugin."}) # fmt: skip + server_url: Optional[str] = field(default=None, metadata={"description": "The plugin server URL used for configuration."}) # fmt: skip + state: Optional[str] = field(default=None, metadata={"description": "The current status of the plugin."}) # fmt: skip + build_status: Optional[str] = field(default=None, metadata={"description": "The status of the plugin."}) # fmt: skip + created_at: Optional[datetime] = field(default=None, metadata={"description": "The timestamp for when the plugin was created."}) # fmt: skip + updated_at: Optional[datetime] = field(default=None, metadata={"description": "The timestamp for when the plugin was last updated."}) # fmt: skip + + @classmethod + def called_collect_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "list-plugins", "plugins"), + AwsApiSpec(service_name, "list-tags-for-resource"), + ] + + @classmethod + def called_mutator_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "tag-resource"), + AwsApiSpec(service_name, "untag-resource"), + AwsApiSpec(service_name, "delete-plugin"), + ] + + def delete_resource(self, client: AwsClient, graph: Graph) -> bool: + client.call( + aws_service="qbusiness", + action="delete-plugin", + result_name=None, + applicationId=self.application_id, + pluginId=self.id, + ) + return True + + +@define(eq=False, slots=False) +class AwsQBusinessRetriever(AmazonQTaggable, AwsResource): + kind: ClassVar[str] = "aws_q_business_retriever" + kind_display: ClassVar[str] = "AWS QBusiness Retriever" + kind_description: ClassVar[str] = ( + "Represents a retriever in the AWS QBusiness service. Retrievers are used to fetch and" + " process data from various sources within the QBusiness ecosystem." + ) + aws_metadata: ClassVar[Dict[str, Any]] = { + "arn_tpl": "arn:{partition}:qbusiness:{region}:{account}:application/{application_id}/retriever/{id}", + "extra_args": ["application_id"], + } + # Collected via AwsQBusinessApplication() + mapping: ClassVar[Dict[str, Bender]] = { + "id": S("retrieverId"), + "tags": S("Tags", default=[]) >> ToDict(), + "name": S("displayName"), + "application_id": S("applicationId"), + "retriever_id": S("retrieverId"), + "type": S("type"), + "status": S("status"), + "display_name": S("displayName"), + } + application_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the Amazon Q Business application using the retriever."}) # fmt: skip + retriever_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the retriever used by your Amazon Q Business application."}) # fmt: skip + type: Optional[str] = field(default=None, metadata={"description": "The type of your retriever."}) # fmt: skip + status: Optional[str] = field(default=None, metadata={"description": "The status of your retriever."}) # fmt: skip + display_name: Optional[str] = field(default=None, metadata={"description": "The name of your retriever."}) # fmt: skip + + @classmethod + def called_collect_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "list-retrievers", "retrievers"), + AwsApiSpec(service_name, "list-tags-for-resource"), + ] + + @classmethod + def called_mutator_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "tag-resource"), + AwsApiSpec(service_name, "untag-resource"), + AwsApiSpec(service_name, "delete-retriever"), + ] + + def delete_resource(self, client: AwsClient, graph: Graph) -> bool: + client.call( + aws_service="qbusiness", + action="delete-retriever", + result_name=None, + applicationId=self.application_id, + retrieverId=self.id, + ) + return True + + +@define(eq=False, slots=False) +class AwsQBusinessWebExperience(AmazonQTaggable, AwsResource): + kind: ClassVar[str] = "aws_q_business_web_experience" + kind_display: ClassVar[str] = "AWS QBusiness Web Experience" + kind_description: ClassVar[str] = ( + "Represents a web experience in the AWS QBusiness service. Web experiences define" + " interactive web-based applications or interfaces within the QBusiness ecosystem." + ) + aws_metadata: ClassVar[Dict[str, Any]] = { + "arn_tpl": "arn:{partition}:qbusiness:{region}:{account}:application/{application_id}/web-experience/{id}", + "extra_args": ["application_id"], + } + # Collected via AwsQBusinessApplication() + mapping: ClassVar[Dict[str, Bender]] = { + "id": S("webExperienceId"), + "tags": S("Tags", default=[]) >> ToDict(), + "name": S("Tags", default=[]) >> TagsValue("Name"), + "ctime": S("createdAt"), + "mtime": S("updatedAt"), + "web_experience_id": S("webExperienceId"), + "created_at": S("createdAt"), + "updated_at": S("updatedAt"), + "default_endpoint": S("defaultEndpoint"), + "status": S("status"), + } + application_id: Optional[str] = field(default=None, metadata={"description": "The identifier of the Amazon Q Business application."}) # fmt: skip + web_experience_id: Optional[str] = field(default=None, metadata={"description": "The identifier of your Amazon Q Business web experience."}) # fmt: skip + created_at: Optional[datetime] = field(default=None, metadata={"description": "The Unix timestamp when the Amazon Q Business application was last updated."}) # fmt: skip + updated_at: Optional[datetime] = field(default=None, metadata={"description": "The Unix timestamp when your Amazon Q Business web experience was updated."}) # fmt: skip + default_endpoint: Optional[str] = field(default=None, metadata={"description": "The endpoint URLs for your Amazon Q Business web experience. The URLs are unique and fully hosted by Amazon Web Services."}) # fmt: skip + status: Optional[str] = field(default=None, metadata={"description": "The status of your Amazon Q Business web experience."}) # fmt: skip + + @classmethod + def called_collect_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "list-web-experiences", "webExperiences"), + AwsApiSpec(service_name, "list-tags-for-resource"), + ] + + @classmethod + def called_mutator_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec(service_name, "tag-resource"), + AwsApiSpec(service_name, "untag-resource"), + AwsApiSpec(service_name, "delete-web-experience"), + ] + + def delete_resource(self, client: AwsClient, graph: Graph) -> bool: + client.call( + aws_service="qbusiness", + action="delete-web-experience", + result_name=None, + applicationId=self.application_id, + webExperienceId=self.id, + ) + return True + + +@define(eq=False, slots=False) +class AwsQAppsCategory: + kind: ClassVar[str] = "aws_q_apps_category" + mapping: ClassVar[Dict[str, Bender]] = {"id": S("id"), "title": S("title")} + id: Optional[str] = field(default=None, metadata={"description": "The unique identifier of the category."}) # fmt: skip + title: Optional[str] = field(default=None, metadata={"description": "The title or name of the category."}) # fmt: skip + + +@define(eq=False, slots=False) +class AwsQAppsLibraryItem(AwsResource): + kind: ClassVar[str] = "aws_q_apps_library_item" + kind_display: ClassVar[str] = "AWS QApps Library Item" + kind_description: ClassVar[str] = ( + "Represents a library item in the AWS QApps service. Library items include resources" + " such as scripts, templates, or other components that can be used in QApps applications." + ) + # Collected via AwsQBusinessApplication() + reference_kinds: ClassVar[ModelReference] = { + "predecessors": {"default": ["aws_q_apps"]}, + } + mapping: ClassVar[Dict[str, Bender]] = { + "id": S("libraryItemId"), + "tags": S("Tags", default=[]) >> ToDict(), + "name": S("Tags", default=[]) >> TagsValue("Name"), + "ctime": S("createdAt"), + "mtime": S("updatedAt"), + "library_item_id": S("libraryItemId"), + "app_id": S("appId"), + "app_version": S("appVersion"), + "library_categories": S("categories", default=[]) >> ForallBend(AwsQAppsCategory.mapping), + "status": S("status"), + "created_at": S("createdAt"), + "created_by": S("createdBy"), + "updated_at": S("updatedAt"), + "updated_by": S("updatedBy"), + "rating_count": S("ratingCount"), + "is_rated_by_user": S("isRatedByUser"), + "user_count": S("userCount"), + } + instance_id: Optional[str] = field(default=None, metadata={"description": "The unique identifier of the environment app."}) # fmt: skip + library_item_id: Optional[str] = field(default=None, metadata={"description": "The unique identifier of the library item."}) # fmt: skip + app_id: Optional[str] = field(default=None, metadata={"description": "The unique identifier of the Q App associated with the library item."}) # fmt: skip + app_version: Optional[int] = field(default=None, metadata={"description": "The version of the Q App associated with the library item."}) # fmt: skip + library_categories: Optional[List[AwsQAppsCategory]] = field(factory=list, metadata={"description": "The categories associated with the library item."}) # fmt: skip + status: Optional[str] = field(default=None, metadata={"description": "The status of the library item."}) # fmt: skip + created_at: Optional[datetime] = field(default=None, metadata={"description": "The date and time the library item was created."}) # fmt: skip + created_by: Optional[str] = field(default=None, metadata={"description": "The user who created the library item."}) # fmt: skip + updated_at: Optional[datetime] = field(default=None, metadata={"description": "The date and time the library item was last updated."}) # fmt: skip + updated_by: Optional[str] = field(default=None, metadata={"description": "The user who last updated the library item."}) # fmt: skip + rating_count: Optional[int] = field(default=None, metadata={"description": "The number of ratings the library item has received."}) # fmt: skip + is_rated_by_user: Optional[bool] = field(default=None, metadata={"description": "Whether the current user has rated the library item."}) # fmt: skip + user_count: Optional[int] = field(default=None, metadata={"description": "The number of users who have the associated Q App."}) # fmt: skip + + def connect_in_graph(self, builder: GraphBuilder, source: Json) -> None: + if app_id := self.app_id: + builder.add_edge(self, reverse=True, clazz=AwsQApps, id=app_id) + + @classmethod + def called_collect_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec("qapps", "list-library-items", "libraryItems"), + AwsApiSpec("qapps", "list-tags-for-resource"), + ] + + @classmethod + def called_mutator_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec("qapps", "tag-resource"), + AwsApiSpec("qapps", "untag-resource"), + AwsApiSpec("qapps", "delete-library-item"), + ] + + def delete_resource(self, client: AwsClient, graph: Graph) -> bool: + client.call( + aws_service="qapps", + action="delete-library-item", + result_name=None, + instanceId=self.instance_id, + libraryItemId=self.id, + ) + return True + + @classmethod + def service_name(cls) -> str: + return "qapps" + + +@define(eq=False, slots=False) +class AwsQApps(AwsResource): + kind: ClassVar[str] = "aws_q_apps" + kind_display: ClassVar[str] = "AWS QApps" + kind_description: ClassVar[str] = ( + "Represents an application within the AWS QApps service. QApps applications include" + " various components and configurations for developing and deploying apps within the AWS environment." + ) + # Collected via AwsQBusinessApplication() + mapping: ClassVar[Dict[str, Bender]] = { + "id": S("appId"), + "tags": S("Tags", default=[]) >> ToDict(), + "name": S("Tags", default=[]) >> TagsValue("Name"), + "app_id": S("appId"), + "app_arn": S("appArn"), + "title": S("title"), + "description": S("description"), + "created_at": S("createdAt"), + "can_edit": S("canEdit"), + "status": S("status"), + "arn": S("appArn"), + } + instance_id: Optional[str] = field(default=None, metadata={"description": "The unique identifier of the environment app."}) # fmt: skip + app_id: Optional[str] = field(default=None, metadata={"description": "The unique identifier of the Q App."}) # fmt: skip + app_arn: Optional[str] = field(default=None, metadata={"description": "The Amazon Resource Name (ARN) of the Q App."}) # fmt: skip + title: Optional[str] = field(default=None, metadata={"description": "The title of the Q App."}) # fmt: skip + description: Optional[str] = field(default=None, metadata={"description": "The description of the Q App."}) # fmt: skip + created_at: Optional[datetime] = field(default=None, metadata={"description": "The date and time the user's association with the Q App was created."}) # fmt: skip + can_edit: Optional[bool] = field(default=None, metadata={"description": "A flag indicating whether the user can edit the Q App."}) # fmt: skip + status: Optional[str] = field(default=None, metadata={"description": "The status of the user's association with the Q App."}) # fmt: skip + + def update_resource_tag(self, client: AwsClient, key: str, value: str) -> bool: + client.call( + aws_service="qapps", + action="tag-resource", + result_name=None, + resourceARN=self.arn, + tags={key: value}, + ) + return True + + def delete_resource_tag(self, client: AwsClient, key: str) -> bool: + client.call( + aws_service="qapps", + action="untag-resource", + result_name=None, + resourceARN=self.arn, + tagKeys=[key], + ) + return True + + @classmethod + def called_collect_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec("qapps", "list-q-apps", "apps"), + AwsApiSpec("qapps", "list-tags-for-resource"), + ] + + @classmethod + def called_mutator_apis(cls) -> List[AwsApiSpec]: + return [ + AwsApiSpec("qapps", "tag-resource"), + AwsApiSpec("qapps", "untag-resource"), + AwsApiSpec("qapps", "delete-q-app"), + ] + + def delete_resource(self, client: AwsClient, graph: Graph) -> bool: + client.call( + aws_service="qapps", + action="delete-q-app", + result_name=None, + instanceId=self.instance_id, + appId=self.id, + ) + return True + + @classmethod + def service_name(cls) -> str: + return "qapps" + + +resources: List[Type[AwsResource]] = [ + AwsQBusinessApplication, + AwsQBusinessConversation, + AwsQBusinessDataSource, + AwsQBusinessDataSourceSyncJob, + AwsQBusinessDocument, + AwsQBusinessIndice, + AwsQBusinessMessage, + AwsQBusinessPlugin, + AwsQBusinessRetriever, + AwsQBusinessWebExperience, + AwsQAppsLibraryItem, + AwsQApps, +] diff --git a/plugins/aws/fix_plugin_aws/resource/backup.py b/plugins/aws/fix_plugin_aws/resource/backup.py index 091ad5b8e3..b3e8f6c4b3 100644 --- a/plugins/aws/fix_plugin_aws/resource/backup.py +++ b/plugins/aws/fix_plugin_aws/resource/backup.py @@ -222,7 +222,7 @@ class AwsBackupAdvancedBackupSetting: @define(eq=False, slots=False) -class AwsBackupPlan(AwsResource, BackupResourceTaggable): +class AwsBackupPlan(BackupResourceTaggable, AwsResource): kind: ClassVar[str] = "aws_backup_plan" kind_display: ClassVar[str] = "AWS Backup Plan" aws_metadata: ClassVar[Dict[str, Any]] = {"provider_link_tpl": "https://{region_id}.console.aws.amazon.com/backup/home?region={region_id}#/backupplan/details/{id}", "arn_tpl": "arn:{partition}:backup:{region}:{account}:backup-plan:{id}"} # fmt: skip @@ -302,7 +302,7 @@ def add_tags(backup_plan: AwsBackupPlan) -> None: @define(eq=False, slots=False) -class AwsBackupVault(AwsResource, BackupResourceTaggable): +class AwsBackupVault(BackupResourceTaggable, AwsResource): kind: ClassVar[str] = "aws_backup_vault" kind_display: ClassVar[str] = "AWS Backup Vault" aws_metadata: ClassVar[Dict[str, Any]] = {"provider_link_tpl": "https://{region_id}.console.aws.amazon.com/backup/home?region={region_id}#/backupplan/details/{name}", "arn_tpl": "arn:{partition}:backup:{region}:{account}:backup-vault:{name}"} # fmt: skip @@ -570,7 +570,7 @@ class AwsBackupReportDeliveryChannel: @define(eq=False, slots=False) -class AwsBackupReportPlan(AwsResource, BackupResourceTaggable): +class AwsBackupReportPlan(BackupResourceTaggable, AwsResource): kind: ClassVar[str] = "aws_backup_report_plan" kind_display: ClassVar[str] = "AWS Report Plan" aws_metadata: ClassVar[Dict[str, Any]] = {"provider_link_tpl": "https://{region_id}.console.aws.amazon.com/backup/home?region={region_id}#/compliance/reports/details/{name}", "arn_tpl": "arn:{partition}:backup:{region}:{account}:report-plan:{name}"} # fmt: skip @@ -658,7 +658,7 @@ def connect_in_graph(self, builder: GraphBuilder, source: Json) -> None: builder.add_edge(self, reverse=True, clazz=AwsBackupFramework, id=framework_arn) -class AwsBackupRestoreTestingPlan(AwsResource, BackupResourceTaggable): +class AwsBackupRestoreTestingPlan(BackupResourceTaggable, AwsResource): kind: ClassVar[str] = "aws_backup_restore_testing_plan" kind_display: ClassVar[str] = "AWS Restore Testing Plan" aws_metadata: ClassVar[Dict[str, Any]] = {"provider_link_tpl": "https://{region_id}.console.aws.amazon.com/backup/home?region={region_id}#/restoretesting/details/{name}", "arn_tpl": "arn:{partition}:backup:{region}:{account}:restore-testing-plan:{name}"} # fmt: skip @@ -736,7 +736,7 @@ def add_tags(restore_plan: AwsBackupRestoreTestingPlan) -> None: @define(eq=False, slots=False) -class AwsBackupLegalHold(AwsResource, BackupResourceTaggable): +class AwsBackupLegalHold(BackupResourceTaggable, AwsResource): kind: ClassVar[str] = "aws_backup_legal_hold" kind_display: ClassVar[str] = "AWS Legal Hold" aws_metadata: ClassVar[Dict[str, Any]] = {"provider_link_tpl": "https://{region_id}.console.aws.amazon.com/backup/home?region={region_id}#/legalholds/details/{id}", "arn_tpl": "arn:{partition}:backup:{region}:{account}:legal-hold:{id}"} # fmt: skip @@ -938,7 +938,7 @@ def connect_in_graph(self, builder: GraphBuilder, source: Json) -> None: @define(eq=False, slots=False) -class AwsBackupFramework(AwsResource, BackupResourceTaggable): +class AwsBackupFramework(BackupResourceTaggable, AwsResource): kind: ClassVar[str] = "aws_backup_framework" kind_display: ClassVar[str] = "AWS Backup Framework" aws_metadata: ClassVar[Dict[str, Any]] = {"provider_link_tpl": "https://{region_id}.console.aws.amazon.com/backup/home?region={region_id}#/compliance/frameworks/details/{name}", "arn_tpl": "arn:{partition}:backup:{region}:{account}:framework:{name}"} # fmt: skip diff --git a/plugins/aws/fix_plugin_aws/resource/base.py b/plugins/aws/fix_plugin_aws/resource/base.py index 084b75a377..a3935e3868 100644 --- a/plugins/aws/fix_plugin_aws/resource/base.py +++ b/plugins/aws/fix_plugin_aws/resource/base.py @@ -508,13 +508,21 @@ def add_node( # if there is no arn: try to create one from template if node.arn is None and (arn_tpl := meta.get("arn_tpl")): try: - node.arn = arn_tpl.format( - partition=self.account.partition, - id=node.id, - name=node.name, - account=self.account.id, - region=self.region.name, - ) + args = { + "partition": self.account.partition, + "id": node.id, + "name": node.name, + "account": self.account.id, + "region": self.region.name, + } + + # Add any additional dynamic arguments from the metadata (if they exist) + if extra_args := meta.get("extra_args"): + for extra_arg in extra_args: + args[extra_arg] = getattr(node, extra_arg) + + # Format the ARN with the provided arguments + node.arn = arn_tpl.format(**args) except Exception as e: log.warning(f"Can not compute ARN for {node} with template: {arn_tpl}: {e}") diff --git a/plugins/aws/test/collector_test.py b/plugins/aws/test/collector_test.py index 58f0f7f318..a2fb676630 100644 --- a/plugins/aws/test/collector_test.py +++ b/plugins/aws/test/collector_test.py @@ -33,8 +33,8 @@ def count_kind(clazz: Type[AwsResource]) -> int: # make sure all threads have been joined assert len(threading.enumerate()) == 1 # ensure the correct number of nodes and edges - assert count_kind(AwsResource) == 236 - assert len(account_collector.graph.edges) == 540 + assert count_kind(AwsResource) == 248 + assert len(account_collector.graph.edges) == 564 assert len(account_collector.graph.deferred_edges) == 2 for node in account_collector.graph.nodes: if isinstance(node, AwsRegion): diff --git a/plugins/aws/test/resources/amazonq_test.py b/plugins/aws/test/resources/amazonq_test.py new file mode 100644 index 0000000000..bcf20952ea --- /dev/null +++ b/plugins/aws/test/resources/amazonq_test.py @@ -0,0 +1,6 @@ +from fix_plugin_aws.resource.amazonq import AwsQBusinessApplication +from test.resources import round_trip_for + + +def test_applications() -> None: + round_trip_for(AwsQBusinessApplication) diff --git a/plugins/aws/test/resources/files/qapps/list-library-items__foo.json b/plugins/aws/test/resources/files/qapps/list-library-items__foo.json new file mode 100644 index 0000000000..91189dc299 --- /dev/null +++ b/plugins/aws/test/resources/files/qapps/list-library-items__foo.json @@ -0,0 +1,32 @@ +{ + "libraryItems": [ + { + "libraryItemId": "foo", + "appId": "foo", + "appVersion": 123, + "categories": [ + { + "id": "foo", + "title": "foo" + }, + { + "id": "foo", + "title": "foo" + }, + { + "id": "foo", + "title": "foo" + } + ], + "status": "foo", + "createdAt": "2024-08-30T12:42:21Z", + "createdBy": "foo", + "updatedAt": "2024-08-30T12:42:21Z", + "updatedBy": "foo", + "ratingCount": 123, + "isRatedByUser": true, + "userCount": 123 + } + ], + "nextToken": "foo" + } \ No newline at end of file diff --git a/plugins/aws/test/resources/files/qapps/list-q-apps__foo.json b/plugins/aws/test/resources/files/qapps/list-q-apps__foo.json new file mode 100644 index 0000000000..fe7724ea57 --- /dev/null +++ b/plugins/aws/test/resources/files/qapps/list-q-apps__foo.json @@ -0,0 +1,15 @@ +{ + "apps": [ + { + "appId": "foo", + "appArn": "foo", + "title": "foo", + "description": "foo", + "createdAt": "2024-08-30T12:42:21Z", + "canEdit": true, + "status": "foo" + } + ], + "nextToken": "foo" + } + \ No newline at end of file diff --git a/plugins/aws/test/resources/files/qbusiness/list-applications.json b/plugins/aws/test/resources/files/qbusiness/list-applications.json new file mode 100644 index 0000000000..37f7fd1ec8 --- /dev/null +++ b/plugins/aws/test/resources/files/qbusiness/list-applications.json @@ -0,0 +1,12 @@ +{ + "nextToken": "foo", + "applications": [ + { + "displayName": "foo", + "applicationId": "foo", + "createdAt": "2024-08-30T12:21:29Z", + "updatedAt": "2024-08-30T12:21:29Z", + "status": "ACTIVE" + } + ] + } \ No newline at end of file diff --git a/plugins/aws/test/resources/files/qbusiness/list-conversations__foo.json b/plugins/aws/test/resources/files/qbusiness/list-conversations__foo.json new file mode 100644 index 0000000000..c79816e22a --- /dev/null +++ b/plugins/aws/test/resources/files/qbusiness/list-conversations__foo.json @@ -0,0 +1,10 @@ +{ + "nextToken": "foo", + "conversations": [ + { + "conversationId": "foo", + "title": "foo", + "startTime": "2024-08-30T12:42:21Z" + } + ] +} \ No newline at end of file diff --git a/plugins/aws/test/resources/files/qbusiness/list-data-source-sync-jobs__foo_foo_foo.json b/plugins/aws/test/resources/files/qbusiness/list-data-source-sync-jobs__foo_foo_foo.json new file mode 100644 index 0000000000..db10222f5a --- /dev/null +++ b/plugins/aws/test/resources/files/qbusiness/list-data-source-sync-jobs__foo_foo_foo.json @@ -0,0 +1,23 @@ +{ + "history": [ + { + "executionId": "foo", + "startTime": "2024-08-30T12:48:51Z", + "endTime": "2024-08-30T12:48:51Z", + "status": "SUCCEEDED", + "error": { + "errorMessage": "foo", + "errorCode": "InvalidRequest" + }, + "dataSourceErrorCode": "foo", + "metrics": { + "documentsAdded": "foo", + "documentsModified": "foo", + "documentsDeleted": "foo", + "documentsFailed": "foo", + "documentsScanned": "foo" + } + } + ], + "nextToken": "foo" + } \ No newline at end of file diff --git a/plugins/aws/test/resources/files/qbusiness/list-data-sources__foo_foo.json b/plugins/aws/test/resources/files/qbusiness/list-data-sources__foo_foo.json new file mode 100644 index 0000000000..a408da4662 --- /dev/null +++ b/plugins/aws/test/resources/files/qbusiness/list-data-sources__foo_foo.json @@ -0,0 +1,13 @@ +{ + "dataSources": [ + { + "displayName": "foo", + "dataSourceId": "foo", + "type": "foo", + "createdAt": "2024-08-30T12:46:23Z", + "updatedAt": "2024-08-30T12:46:23Z", + "status": "CREATING" + } + ], + "nextToken": "foo" + } \ No newline at end of file diff --git a/plugins/aws/test/resources/files/qbusiness/list-documents__foo_foo.json b/plugins/aws/test/resources/files/qbusiness/list-documents__foo_foo.json new file mode 100644 index 0000000000..5b1adb080a --- /dev/null +++ b/plugins/aws/test/resources/files/qbusiness/list-documents__foo_foo.json @@ -0,0 +1,15 @@ +{ + "documentDetailList": [ + { + "documentId": "foo", + "status": "PROCESSING", + "error": { + "errorMessage": "foo", + "errorCode": "InvalidRequest" + }, + "createdAt": "2024-08-30T12:46:23Z", + "updatedAt": "2024-08-30T12:46:23Z" + } + ], + "nextToken": "foo" + } \ No newline at end of file diff --git a/plugins/aws/test/resources/files/qbusiness/list-indices__foo.json b/plugins/aws/test/resources/files/qbusiness/list-indices__foo.json new file mode 100644 index 0000000000..75e0dfd3f6 --- /dev/null +++ b/plugins/aws/test/resources/files/qbusiness/list-indices__foo.json @@ -0,0 +1,12 @@ +{ + "nextToken": "foo", + "indices": [ + { + "displayName": "foo", + "indexId": "foo", + "createdAt": "2024-08-30T12:42:21Z", + "updatedAt": "2024-08-30T12:42:21Z", + "status": "ACTIVE" + } + ] + } \ No newline at end of file diff --git a/plugins/aws/test/resources/files/qbusiness/list-messages__foo_foo.json b/plugins/aws/test/resources/files/qbusiness/list-messages__foo_foo.json new file mode 100644 index 0000000000..08bd1bc813 --- /dev/null +++ b/plugins/aws/test/resources/files/qbusiness/list-messages__foo_foo.json @@ -0,0 +1,220 @@ +{ + "messages": [ + { + "messageId": "foo", + "body": "foo", + "time": "2024-08-30T12:46:23Z", + "type": "SYSTEM", + "attachments": [ + { + "name": "foo", + "status": "SUCCEEDED", + "error": { + "errorMessage": "foo", + "errorCode": "InvalidRequest" + } + }, + { + "name": "foo", + "status": "SUCCEEDED", + "error": { + "errorMessage": "foo", + "errorCode": "InvalidRequest" + } + }, + { + "name": "foo", + "status": "SUCCEEDED", + "error": { + "errorMessage": "foo", + "errorCode": "InvalidRequest" + } + } + ], + "sourceAttribution": [ + { + "title": "foo", + "snippet": "foo", + "url": "https://example.com", + "citationNumber": 123, + "updatedAt": "2024-08-30T12:46:23Z", + "textMessageSegments": [ + { + "beginOffset": 123, + "endOffset": 123, + "snippetExcerpt": { + "text": "foo" + } + }, + { + "beginOffset": 123, + "endOffset": 123, + "snippetExcerpt": { + "text": "foo" + } + }, + { + "beginOffset": 123, + "endOffset": 123, + "snippetExcerpt": { + "text": "foo" + } + } + ] + }, + { + "title": "foo", + "snippet": "foo", + "url": "https://example.com", + "citationNumber": 123, + "updatedAt": "2024-08-30T12:46:23Z", + "textMessageSegments": [ + { + "beginOffset": 123, + "endOffset": 123, + "snippetExcerpt": { + "text": "foo" + } + }, + { + "beginOffset": 123, + "endOffset": 123, + "snippetExcerpt": { + "text": "foo" + } + }, + { + "beginOffset": 123, + "endOffset": 123, + "snippetExcerpt": { + "text": "foo" + } + } + ] + }, + { + "title": "foo", + "snippet": "foo", + "url": "https://example.com", + "citationNumber": 123, + "updatedAt": "2024-08-30T12:46:23Z", + "textMessageSegments": [ + { + "beginOffset": 123, + "endOffset": 123, + "snippetExcerpt": { + "text": "foo" + } + }, + { + "beginOffset": 123, + "endOffset": 123, + "snippetExcerpt": { + "text": "foo" + } + }, + { + "beginOffset": 123, + "endOffset": 123, + "snippetExcerpt": { + "text": "foo" + } + } + ] + } + ], + "actionReview": { + "pluginId": "foo", + "pluginType": "SALESFORCE", + "payload": { + "0": { + "displayName": "foo", + "displayOrder": 123, + "displayDescription": "foo", + "type": "NUMBER", + "value": {}, + "allowedValues": [ + { + "value": {}, + "displayValue": {} + }, + { + "value": {}, + "displayValue": {} + }, + { + "value": {}, + "displayValue": {} + } + ], + "allowedFormat": "foo", + "required": true + }, + "1": { + "displayName": "foo", + "displayOrder": 123, + "displayDescription": "foo", + "type": "NUMBER", + "value": {}, + "allowedValues": [ + { + "value": {}, + "displayValue": {} + }, + { + "value": {}, + "displayValue": {} + }, + { + "value": {}, + "displayValue": {} + } + ], + "allowedFormat": "foo", + "required": true + }, + "2": { + "displayName": "foo", + "displayOrder": 123, + "displayDescription": "foo", + "type": "NUMBER", + "value": {}, + "allowedValues": [ + { + "value": {}, + "displayValue": {} + }, + { + "value": {}, + "displayValue": {} + }, + { + "value": {}, + "displayValue": {} + } + ], + "allowedFormat": "foo", + "required": true + } + }, + "payloadFieldNameSeparator": "foo" + }, + "actionExecution": { + "pluginId": "foo", + "payload": { + "0": { + "value": {} + }, + "1": { + "value": {} + }, + "2": { + "value": {} + } + }, + "payloadFieldNameSeparator": "foo" + } + } + ], + "nextToken": "foo" + } \ No newline at end of file diff --git a/plugins/aws/test/resources/files/qbusiness/list-plugins__foo.json b/plugins/aws/test/resources/files/qbusiness/list-plugins__foo.json new file mode 100644 index 0000000000..34367dcdc2 --- /dev/null +++ b/plugins/aws/test/resources/files/qbusiness/list-plugins__foo.json @@ -0,0 +1,15 @@ +{ + "nextToken": "foo", + "plugins": [ + { + "pluginId": "foo", + "displayName": "foo", + "type": "SALESFORCE", + "serverUrl": "https://example.com", + "state": "DISABLED", + "buildStatus": "CREATE_IN_PROGRESS", + "createdAt": "2024-08-30T12:42:21Z", + "updatedAt": "2024-08-30T12:42:21Z" + } + ] + } \ No newline at end of file diff --git a/plugins/aws/test/resources/files/qbusiness/list-retrievers__foo.json b/plugins/aws/test/resources/files/qbusiness/list-retrievers__foo.json new file mode 100644 index 0000000000..91fdcd66a7 --- /dev/null +++ b/plugins/aws/test/resources/files/qbusiness/list-retrievers__foo.json @@ -0,0 +1,12 @@ +{ + "retrievers": [ + { + "applicationId": "foo", + "retrieverId": "foo", + "type": "KENDRA_INDEX", + "status": "ACTIVE", + "displayName": "foo" + } + ], + "nextToken": "foo" + } \ No newline at end of file diff --git a/plugins/aws/test/resources/files/qbusiness/list-web-experiences__foo.json b/plugins/aws/test/resources/files/qbusiness/list-web-experiences__foo.json new file mode 100644 index 0000000000..ec0b657f12 --- /dev/null +++ b/plugins/aws/test/resources/files/qbusiness/list-web-experiences__foo.json @@ -0,0 +1,12 @@ +{ + "webExperiences": [ + { + "webExperienceId": "foo", + "createdAt": "2024-08-30T12:42:21Z", + "updatedAt": "2024-08-30T12:42:21Z", + "defaultEndpoint": "https://example.com", + "status": "ACTIVE" + } + ], + "nextToken": "foo" + } \ No newline at end of file diff --git a/plugins/aws/tools/aws_model_gen.py b/plugins/aws/tools/aws_model_gen.py index b97cddedd5..2e4d6c59a8 100644 --- a/plugins/aws/tools/aws_model_gen.py +++ b/plugins/aws/tools/aws_model_gen.py @@ -943,6 +943,22 @@ def default_imports() -> str: "wafv2": [ # AwsFixModel("get-logging-configuration", "LoggingConfigurations", "LoggingConfiguration", prefix="Waf") ], + "qbusiness": [ + AwsFixModel( + api_action="list-applications", + result_property="applications", + result_shape="Applications", + prefix="QBusiness", + ), + ], + "qapps": [ + AwsFixModel( + api_action="list-qapps", + result_property="apps", + result_shape="ListQAppsOutput", + prefix="QApps", + ), + ], "backup": [ # AwsFixModel( # api_action="list-backup-job-summaries", @@ -950,39 +966,6 @@ def default_imports() -> str: # result_shape="BackupJobSummaryList", # prefix="Backup", # ), - # AwsFixModel( - # api_action="list-backup-jobs", result_property="BackupJobs", result_shape="BackupJobsList", prefix="Backup" - # ), - # AwsFixModel( - # api_action="list-backup-plan-templates", - # result_property="BackupPlanTemplates", - # result_shape="BackupPlanTemplatesList", - # prefix="Backup", - # ), - # AwsFixModel( - # api_action="list-backup-plan-versions", - # result_property="BackupPlanVersions", - # result_shape="BackupPlanVersionsList", - # prefix="Backup", - # ), - # AwsFixModel( - # api_action="list-backup-plans", - # result_property="BackupPlans", - # result_shape="BackupPlansList", - # prefix="Backup", - # ), - # AwsFixModel( - # api_action="list-backup-selections", - # result_property="BackupSelections", - # result_shape="BackupSelectionsList", - # prefix="Backup", - # ), - # AwsFixModel( - # api_action="list-backup-vaults", - # result_property="BackupVaultList", - # result_shape="BackupVaultListMember", - # prefix="Backup", - # ), ], }