How to use _add_topics method in localstack

Best Python code snippet using localstack_python

sns.py

Source:sns.py

...55            return result56        def _topic_arn(params, resources, resource_id, **kwargs):57            resource = cls(resources[resource_id])58            return resource.physical_resource_id or resource.get_physical_resource_id()59        def _add_topics(resource_id, resources, resource_type, func, stack_name):60            sns = aws_stack.connect_to_service("sns")61            resource = cls(resources[resource_id])62            props = resource.props63            subscriptions = props.get("Subscription", [])64            for subscription in subscriptions:65                if is_none_or_empty_value(subscription):66                    continue67                endpoint = subscription["Endpoint"]68                topic_arn = retrieve_topic_arn(sns, props["TopicName"])69                sns.subscribe(70                    TopicArn=topic_arn, Protocol=subscription["Protocol"], Endpoint=endpoint71                )72        return {73            "create": [...

topic_info.py

Source:topic_info.py

...39        """Run the post-initialization."""40        self.creation_date = datetime.datetime.now()41        self.path = pathlib.Path(self.path).resolve()42    @staticmethod43    def _add_topics(44        mapping: dict[str, list[str]], kind: str, new_topics: Iterable[str]45    ) -> None:46        """Add topics to a mapping with collection of topics.47        Parameters48        ----------49        mapping50            A mapping of the form kind -> list-of-topics that shall be51            updated in-place. For example ``{"MeSH": ["topic 1", "topic 2"]}``.52        kind53            The topic kind. Corresponds to a key in ``mapping``.54        new_topics55            The topics to add. Corresponds to a value in ``mapping``.56        """57        topics = mapping.get(kind, [])58        topics.extend(new_topics)59        mapping[kind] = sorted(set(topics))60    def add_article_topics(self, kind: str, topics: Iterable[str]) -> None:61        """Add article topics.62        Parameters63        ----------64        kind65            The topic kind. For example "MeSH" or "MAG".66        topics67            A collection of the topics to add.68        """69        self._add_topics(self.article_topics, kind, topics)70    def add_journal_topics(self, kind: str, topics: Iterable[str]) -> None:71        """Add journal topics.72        Parameters73        ----------74        kind75            The topic kind. For example "MeSH" or "MAG".76        topics77            A collection of the topics to add.78        """79        self._add_topics(self.journal_topics, kind, topics)80    def json(self) -> dict:81        """Convert the contents of this class to a structured dictionary.82        Apart from the source, path and topic entries a "metadata" top-level83        key will be added containing a dictionary with entries "created-date"84        and "bbs-version".85        Returns86        -------87        dict88            The structure dictionary with all topic information.89        """90        metadata: dict[str, Any] = {91            "created-date": self.creation_date.strftime("%Y-%m-%d %H:%M:%S"),92            "bbs-version": bluesearch.__version__,93        }...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.