KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _next_sync method in localstack

Best Python code snippet using localstack_python

asgi.py

Source:asgi.py

...69    :return: an async generator70    """71    loop = loop or asyncio.get_event_loop()72    stop = object()73    def _next_sync():74        try:75            # this call may potentially call blocking IO, which is why we call it in an executor76            return next(it)77        except StopIteration:78            return stop79    while True:80        val = await loop.run_in_executor(executor, _next_sync)81        if val is stop:82            return83        yield val84class HTTPRequestEventStreamAdapter:85    """86    An adapter to expose an ASGIReceiveCallable coroutine that returns HTTPRequestEvent87    instances, as a PEP 3333 InputStream for consumption in synchronous WSGI/Werkzeug code....

active_sync_status_dto_model.py

Source:active_sync_status_dto_model.py

1# coding: utf-82"""3    Grafana HTTP API.4    The Grafana backend exposes an HTTP API, the same API is used by the frontend to do everything from saving dashboards, creating users and updating data sources.  # noqa: E5015    OpenAPI spec version: 0.0.16    Contact: hello@grafana.com7    Generated by: https://github.com/swagger-api/swagger-codegen.git8"""9import pprint10import re  # noqa: F40111import six12from gpyclient.configuration import Configuration13class ActiveSyncStatusDTOModel(object):14    """NOTE: This class is auto generated by the swagger code generator program.15    Do not edit the class manually.16    """17    """18    Attributes:19      swagger_types (dict): The key is attribute name20                            and the value is attribute type.21      attribute_map (dict): The key is attribute name22                            and the value is json key in definition.23    """24    swagger_types = {25        'enabled': 'bool',26        'next_sync': 'datetime',27        'prev_sync': 'SyncResultModel',28        'schedule': 'str'29    }30    attribute_map = {31        'enabled': 'enabled',32        'next_sync': 'nextSync',33        'prev_sync': 'prevSync',34        'schedule': 'schedule'35    }36    def __init__(self, enabled=None, next_sync=None, prev_sync=None, schedule=None, _configuration=None):  # noqa: E50137        """ActiveSyncStatusDTOModel - a model defined in Swagger"""  # noqa: E50138        if _configuration is None:39            _configuration = Configuration()40        self._configuration = _configuration41        self._enabled = None42        self._next_sync = None43        self._prev_sync = None44        self._schedule = None45        self.discriminator = None46        if enabled is not None:47            self.enabled = enabled48        if next_sync is not None:49            self.next_sync = next_sync50        if prev_sync is not None:51            self.prev_sync = prev_sync52        if schedule is not None:53            self.schedule = schedule54    @property55    def enabled(self):56        """Gets the enabled of this ActiveSyncStatusDTOModel.  # noqa: E50157        :return: The enabled of this ActiveSyncStatusDTOModel.  # noqa: E50158        :rtype: bool59        """60        return self._enabled61    @enabled.setter62    def enabled(self, enabled):63        """Sets the enabled of this ActiveSyncStatusDTOModel.64        :param enabled: The enabled of this ActiveSyncStatusDTOModel.  # noqa: E50165        :type: bool66        """67        self._enabled = enabled68    @property69    def next_sync(self):70        """Gets the next_sync of this ActiveSyncStatusDTOModel.  # noqa: E50171        :return: The next_sync of this ActiveSyncStatusDTOModel.  # noqa: E50172        :rtype: datetime73        """74        return self._next_sync75    @next_sync.setter76    def next_sync(self, next_sync):77        """Sets the next_sync of this ActiveSyncStatusDTOModel.78        :param next_sync: The next_sync of this ActiveSyncStatusDTOModel.  # noqa: E50179        :type: datetime80        """81        self._next_sync = next_sync82    @property83    def prev_sync(self):84        """Gets the prev_sync of this ActiveSyncStatusDTOModel.  # noqa: E50185        :return: The prev_sync of this ActiveSyncStatusDTOModel.  # noqa: E50186        :rtype: SyncResultModel87        """88        return self._prev_sync89    @prev_sync.setter90    def prev_sync(self, prev_sync):91        """Sets the prev_sync of this ActiveSyncStatusDTOModel.92        :param prev_sync: The prev_sync of this ActiveSyncStatusDTOModel.  # noqa: E50193        :type: SyncResultModel94        """95        self._prev_sync = prev_sync96    @property97    def schedule(self):98        """Gets the schedule of this ActiveSyncStatusDTOModel.  # noqa: E50199        :return: The schedule of this ActiveSyncStatusDTOModel.  # noqa: E501100        :rtype: str101        """102        return self._schedule103    @schedule.setter104    def schedule(self, schedule):105        """Sets the schedule of this ActiveSyncStatusDTOModel.106        :param schedule: The schedule of this ActiveSyncStatusDTOModel.  # noqa: E501107        :type: str108        """109        self._schedule = schedule110    def to_dict(self):111        """Returns the model properties as a dict"""112        result = {}113        for attr, _ in six.iteritems(self.swagger_types):114            value = getattr(self, attr)115            if isinstance(value, list):116                result[attr] = list(map(117                    lambda x: x.to_dict() if hasattr(x, "to_dict") else x,118                    value119                ))120            elif hasattr(value, "to_dict"):121                result[attr] = value.to_dict()122            elif isinstance(value, dict):123                result[attr] = dict(map(124                    lambda item: (item[0], item[1].to_dict())125                    if hasattr(item[1], "to_dict") else item,126                    value.items()127                ))128            else:129                result[attr] = value130        if issubclass(ActiveSyncStatusDTOModel, dict):131            for key, value in self.items():132                result[key] = value133        return result134    def to_str(self):135        """Returns the string representation of the model"""136        return pprint.pformat(self.to_dict())137    def __repr__(self):138        """For `print` and `pprint`"""139        return self.to_str()140    def __eq__(self, other):141        """Returns true if both objects are equal"""142        if not isinstance(other, ActiveSyncStatusDTOModel):143            return False144        return self.to_dict() == other.to_dict()145    def __ne__(self, other):146        """Returns true if both objects are not equal"""147        if not isinstance(other, ActiveSyncStatusDTOModel):148            return True...

client.py

Source:client.py

1from time import monotonic2from threading import Semaphore3from typing import Callable, Dict, TYPE_CHECKING4from flypper.context import Context5if TYPE_CHECKING:6    from flypper.entities.flag import Flag7    from flypper.storage.abstract import AbstractStorage8class Client:9    """Client caches the flags' configuration at the application level.10    A client set a TTL: the maximum duration it can serve flags without synchronizing11    with the storage backend.12    It efficiently synchronize by only asking for the updates since its last sync.13    To get there, the storage and the client agree on a global version number associated14    with each update.15    """16    def __init__(17        self,18        storage: "AbstractStorage",19        ttl: float = 5.0,20        time_fn: Callable[[], float] = monotonic,21    ):22        self._storage: "AbstractStorage" = storage23        self._ttl: float = ttl24        self._last_version: int = 025        self._next_sync: float = 026        self._flags: Dict[str, "Flag"] = {}27        self._time_fn: Callable[[], float] = time_fn28        self._semaphore: Semaphore = Semaphore()29    def flags(self) -> Dict[str, "Flag"]:30        """Lists the flag, by their name.31        It will call the storage at most once every ttl seconds, fetching32        only the latest updates since the last storage roundtrip.33        """34        self._sync()35        return self._flags36    def __call__(self, **entries: str) -> Context:37        """Builds a context from this client."""38        return Context(client=self, entries=entries)39    def _sync(self):40        """Syncs the cache with the storage."""41        with self._semaphore:42            now = self._time_fn()43            if now < self._next_sync:44                return45            # Get the latest flags updates from the backend.46            new_flags = self._storage.list(version__gt=self._last_version)47            if new_flags:48                # Don't directly update the _flags, create a copy so the running contexts49                # can operate using the reference of the outdated copy they might have.50                self._flags = self._flags.copy()51                # Update the cached flags with their latest version.52                for new_flag in new_flags:53                    if new_flag.is_deleted:54                        del self._flags[new_flag.name]55                    else:56                        self._flags[new_flag.name] = new_flag57                # Keep track of the latest version we received.58                self._last_version = max(flag.version for flag in new_flags)59            # Compute the minimum time the next sync could occur....

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.