How to use get_models method in localstack

Best Python code snippet using localstack_python

instance.py

Source:instance.py

...84    :param attr_name:85    :return:86    """87    def _getter(self):88        return get_value_or_default(attr_name, self.get_models(), self.name)89    return property(_getter)90class Instance(object):91    name = None  # type: Text92    _sockets = None  # type: Deque[Tuple[zmq.Socket, float]]93    def __init__(94        self,95        context,  # type: zmq.Context96        name,  # type: Text97        zmq_socket,  # type: Text98        street_network_configurations,99        ridesharing_configurations,100        realtime_proxies_configuration,101        zmq_socket_type,102        autocomplete_type,103        instance_equipment_providers,  # type: List[Text]104        streetnetwork_backend_manager,105        external_service_provider_configurations,106        pt_planners_configurations,107        instance_db=None,108    ):109        self.geom = None110        self.geojson = None111        self._sockets = deque()112        self.socket_path = zmq_socket113        self._scenario = None114        self._scenario_name = None115        self.lock = Lock()116        self.context = context117        self.name = name118        self.timezone = None  # timezone will be fetched from the kraken119        self.publication_date = -1120        self.is_initialized = False  # kraken hasn't been called yet we don't have geom nor timezone121        self.breaker = pybreaker.CircuitBreaker(122            fail_max=app.config.get(str('CIRCUIT_BREAKER_MAX_INSTANCE_FAIL'), 5),123            reset_timeout=app.config.get(str('CIRCUIT_BREAKER_INSTANCE_TIMEOUT_S'), 60),124        )125        self.georef = georef.Kraken(self)126        self._streetnetwork_backend_manager = streetnetwork_backend_manager127        disable_database = app.config[str('DISABLE_DATABASE')]128        if disable_database:129            self._streetnetwork_backend_manager.init_streetnetwork_backends_legacy(130                self, street_network_configurations131            )132            self.ridesharing_services_manager = RidesharingServiceManager(self, ridesharing_configurations)133        else:134            self.ridesharing_services_manager = RidesharingServiceManager(135                self, ridesharing_configurations, self.get_ridesharing_services_from_db136            )137        self._pt_planner_manager = pt_planners_manager.PtPlannersManager(138            pt_planners_configurations,139            name,140            zmq_socket_type,141            context,142            default_socket_path=zmq_socket,143            db_configs_getter=lambda: self.pt_planners_configurations,144        )145        # Init Ridesharing services from config file146        self.ridesharing_services_manager.init_ridesharing_services()147        self.ptref = ptref.PtRef(self)148        self.schedule = schedule.MixedSchedule(self)149        providers_getter = None if disable_database else self.get_realtime_proxies_from_db150        self.realtime_proxy_manager = realtime_schedule.RealtimeProxyManager(151            realtime_proxies_configuration, self, providers_getter152        )153        self._autocomplete_type = autocomplete_type154        if self._autocomplete_type is not None and self._autocomplete_type not in global_autocomplete:155            raise RuntimeError(156                'impossible to find autocomplete system {} '157                'cannot initialize instance {}'.format(autocomplete_type, name)158            )159        self.zmq_socket_type = zmq_socket_type160        if disable_database:161            self.equipment_provider_manager = EquipmentProviderManager(162                app.config[str('EQUIPMENT_DETAILS_PROVIDERS')]163            )164        else:165            self.equipment_provider_manager = EquipmentProviderManager(166                app.config[str('EQUIPMENT_DETAILS_PROVIDERS')], self.get_providers_from_db167            )168        # Init equipment providers from config169        self.equipment_provider_manager.init_providers(instance_equipment_providers)170        # Init free-floating providers from config171        if disable_database:172            self.external_service_provider_manager = ExternalServiceManager(173                self, external_service_provider_configurations174            )175        else:176            self.external_service_provider_manager = ExternalServiceManager(177                self, external_service_provider_configurations, self.get_external_service_providers_from_db178            )179        self.external_service_provider_manager.init_external_services()180        self.instance_db = instance_db181    def get_providers_from_db(self):182        """183        :return: a callable query of equipment providers associated to the current instance in db184        """185        models = self._get_models()186        return models.equipment_details_providers if models else None187    def get_ridesharing_services_from_db(self):188        """189        :return: a callable query of ridesharing services associated to the current instance in db190        """191        models = self._get_models()192        return models.ridesharing_services if models else None193    def get_external_service_providers_from_db(self):194        """195        :return: a callable query of external services associated to the current instance in db196        """197        models = self._get_models()198        result = models.external_services if models else None199        return [200            res201            for res in result202            if res.navitia_service in ['free_floatings', 'vehicle_occupancies', "vehicle_positions"]203        ]204    def get_realtime_proxies_from_db(self):205        """206        :return: a callable query of external services associated to the current instance in db207        """208        models = self._get_models()209        result = models.external_services if models else None210        return [res for res in result if res.navitia_service == 'realtime_proxies']211    @property212    def autocomplete(self):213        if self._autocomplete_type:214            # retrocompat: we need to continue to read configuration from file215            # while we migrate to database configuration216            return global_autocomplete.get(self._autocomplete_type)217        backend = global_autocomplete.get(self.autocomplete_backend)218        if backend is None:219            raise RuntimeError(220                'impossible to find autocomplete {} for instance {}'.format(self.autocomplete_backend, self.name)221            )222        return backend223    def stop_point_fallbacks(self):224        return [a for a in global_autocomplete.values() if a.is_handling_stop_points()]225    def get_models(self):226        if self.name not in g.instances_model:227            g.instances_model[self.name] = self._get_models()228        return g.instances_model[self.name]229    def __repr__(self):230        return 'instance.{}'.format(self.name)231    @memory_cache.memoize(app.config[str('MEMORY_CACHE_CONFIGURATION')].get(str('TIMEOUT_PARAMS'), 30))232    @cache.memoize(app.config[str('CACHE_CONFIGURATION')].get(str('TIMEOUT_PARAMS'), 300))233    def _get_models(self):234        if app.config['DISABLE_DATABASE']:235            return None236        if not can_connect_to_database():237            return self.instance_db238        try:239            instance_db = models.Instance.get_by_name(self.name)240        except Exception as e:241            logging.getLogger(__name__).error('No access to table instance (error: {})'.format(e))242            g.can_connect_to_database = False243            return self.instance_db244        self.instance_db = instance_db245        return self.instance_db246    def scenario(self, override_scenario=None):247        """248        once a scenario has been chosen for a request for an instance (coverage), we cannot change it249        """250        if hasattr(g, 'scenario') and g.scenario.get(self.name):251            return g.scenario[self.name]252        def replace_experimental_scenario(s):253            return 'distributed' if s == 'experimental' else s254        if override_scenario:255            logger = logging.getLogger(__name__)256            logger.debug('overriding the scenario for %s with %s', self.name, override_scenario)257            try:258                # for the sake of backwards compatibility... some users may still be using experimental...259                override_scenario = replace_experimental_scenario(override_scenario)260                module = import_module('jormungandr.scenarios.{}'.format(override_scenario))261            except ImportError:262                logger.exception('scenario not found')263                abort(404, message='invalid scenario: {}'.format(override_scenario))264            scenario = module.Scenario()265            # Save scenario_name and scenario266            self._scenario_name = override_scenario267            self._scenario = scenario268            if not hasattr(g, 'scenario'):269                g.scenario = {}270            g.scenario[self.name] = scenario271            return scenario272        instance_db = self.get_models()273        scenario_name = instance_db.scenario if instance_db else 'new_default'274        # for the sake of backwards compatibility... some users may still be using experimental...275        scenario_name = replace_experimental_scenario(scenario_name)276        if not self._scenario or scenario_name != self._scenario_name:277            logger = logging.getLogger(__name__)278            logger.info('loading of scenario %s for instance %s', scenario_name, self.name)279            self._scenario_name = scenario_name280            module = import_module('jormungandr.scenarios.{}'.format(scenario_name))281            self._scenario = module.Scenario()282        # we save the used scenario for future use283        if not hasattr(g, 'scenario'):284            g.scenario = {}285        g.scenario[self.name] = self._scenario286        return self._scenario287    @property288    def max_waiting_duration(self):289        # type: () -> int290        instance_db = self.get_models()291        return get_value_or_default('max_waiting_duration', instance_db, self.name)292    @property293    def journey_order(self):294        # type: () -> Text295        instance_db = self.get_models()296        return get_value_or_default('journey_order', instance_db, self.name)297    @property298    def autocomplete_backend(self):299        # type: () -> Text300        instance_db = self.get_models()301        return get_value_or_default('autocomplete_backend', instance_db, self.name)302    @property303    def max_walking_duration_to_pt(self):304        # type: () -> int305        instance_db = self.get_models()306        return get_value_or_default('max_walking_duration_to_pt', instance_db, self.name)307    @property308    def max_bss_duration_to_pt(self):309        # type: () -> int310        instance_db = self.get_models()311        return get_value_or_default('max_bss_duration_to_pt', instance_db, self.name)312    @property313    def max_bike_duration_to_pt(self):314        # type: () -> int315        instance_db = self.get_models()316        return get_value_or_default('max_bike_duration_to_pt', instance_db, self.name)317    @property318    def max_car_duration_to_pt(self):319        # type: () -> int320        instance_db = self.get_models()321        return get_value_or_default('max_car_duration_to_pt', instance_db, self.name)322    @property323    def max_car_no_park_duration_to_pt(self):324        # type: () -> int325        instance_db = self.get_models()326        return get_value_or_default('max_car_no_park_duration_to_pt', instance_db, self.name)327    @property328    def max_ridesharing_duration_to_pt(self):329        # type: () -> int330        instance_db = self.get_models()331        return get_value_or_default('max_ridesharing_duration_to_pt', instance_db, self.name)332    @property333    def max_taxi_duration_to_pt(self):334        # type: () -> int335        instance_db = self.get_models()336        return get_value_or_default('max_taxi_duration_to_pt', instance_db, self.name)337    @property338    def walking_speed(self):339        # type: () -> float340        instance_db = self.get_models()341        return get_value_or_default('walking_speed', instance_db, self.name)342    @property343    def bss_speed(self):344        # type: () -> float345        instance_db = self.get_models()346        return get_value_or_default('bss_speed', instance_db, self.name)347    @property348    def bike_speed(self):349        # type: () -> float350        instance_db = self.get_models()351        return get_value_or_default('bike_speed', instance_db, self.name)352    @property353    def car_speed(self):354        # type: () -> float355        instance_db = self.get_models()356        return get_value_or_default('car_speed', instance_db, self.name)357    @property358    def car_no_park_speed(self):359        # type: () -> float360        instance_db = self.get_models()361        return get_value_or_default('car_no_park_speed', instance_db, self.name)362    @property363    def ridesharing_speed(self):364        # type: () -> float365        instance_db = self.get_models()366        return get_value_or_default('ridesharing_speed', instance_db, self.name)367    @property368    def max_nb_transfers(self):369        # type: () -> int370        instance_db = self.get_models()371        return get_value_or_default('max_nb_transfers', instance_db, self.name)372    @property373    def min_bike(self):374        # type: () -> int375        instance_db = self.get_models()376        return get_value_or_default('min_bike', instance_db, self.name)377    @property378    def min_bss(self):379        # type: () -> int380        instance_db = self.get_models()381        return get_value_or_default('min_bss', instance_db, self.name)382    @property383    def min_car(self):384        # type: () -> int385        instance_db = self.get_models()386        return get_value_or_default('min_car', instance_db, self.name)387    @property388    def min_taxi(self):389        # type: () -> int390        instance_db = self.get_models()391        return get_value_or_default('min_taxi', instance_db, self.name)392    @property393    def successive_physical_mode_to_limit_id(self):394        # type: () -> Text395        instance_db = self.get_models()396        return get_value_or_default('successive_physical_mode_to_limit_id', instance_db, self.name)397    @property398    def priority(self):399        # type: () -> int400        instance_db = self.get_models()401        return get_value_or_default('priority', instance_db, self.name)402    @property403    def bss_provider(self):404        # type: () -> bool405        instance_db = self.get_models()406        return get_value_or_default('bss_provider', instance_db, self.name)407    @property408    def car_park_provider(self):409        # type: () -> bool410        instance_db = self.get_models()411        return get_value_or_default('car_park_provider', instance_db, self.name)412    @property413    def max_additional_connections(self):414        # type: () -> int415        instance_db = self.get_models()416        return get_value_or_default('max_additional_connections', instance_db, self.name)417    @property418    def is_free(self):419        # type: () -> bool420        instance_db = self.get_models()421        if not instance_db:422            return False423        else:424            return instance_db.is_free425    @property426    def is_open_data(self):427        # type: () -> bool428        instance_db = self.get_models()429        if not instance_db:430            return False431        else:432            return instance_db.is_open_data433    @property434    def max_duration(self):435        # type: () -> int436        instance_db = self.get_models()437        return get_value_or_default('max_duration', instance_db, self.name)438    @property439    def walking_transfer_penalty(self):440        # type: () -> int441        instance_db = self.get_models()442        return get_value_or_default('walking_transfer_penalty', instance_db, self.name)443    @property444    def arrival_transfer_penalty(self):445        # type: () -> int446        instance_db = self.get_models()447        return get_value_or_default('arrival_transfer_penalty', instance_db, self.name)448    @property449    def night_bus_filter_max_factor(self):450        # type: () -> int451        instance_db = self.get_models()452        return get_value_or_default('night_bus_filter_max_factor', instance_db, self.name)453    @property454    def night_bus_filter_base_factor(self):455        # type: () -> float456        instance_db = self.get_models()457        return get_value_or_default('night_bus_filter_base_factor', instance_db, self.name)458    @property459    def realtime_pool_size(self):460        # type: () -> int461        instance_db = self.get_models()462        return get_value_or_default('realtime_pool_size', instance_db, self.name)463    @property464    def min_nb_journeys(self):465        # type: () -> int466        instance_db = self.get_models()467        return get_value_or_default('min_nb_journeys', instance_db, self.name)468    @property469    def max_nb_journeys(self):470        # type: () -> int471        instance_db = self.get_models()472        return get_value_or_default('max_nb_journeys', instance_db, self.name)473    @property474    def min_journeys_calls(self):475        # type: () -> int476        instance_db = self.get_models()477        return get_value_or_default('min_journeys_calls', instance_db, self.name)478    @property479    def max_successive_physical_mode(self):480        # type: () -> int481        instance_db = self.get_models()482        return get_value_or_default('max_successive_physical_mode', instance_db, self.name)483    @property484    def final_line_filter(self):485        instance_db = self.get_models()486        return get_value_or_default('final_line_filter', instance_db, self.name)487    @property488    def max_extra_second_pass(self):489        # type: () -> int490        instance_db = self.get_models()491        return get_value_or_default('max_extra_second_pass', instance_db, self.name)492    @property493    def asynchronous_ridesharing(self):494        # type: () -> bool495        instance_db = self.get_models()496        return get_value_or_default('asynchronous_ridesharing', instance_db, self.name)497    @property498    def ridesharing_greenlet_pool_size(self):499        # type: () -> int500        instance_db = self.get_models()501        return get_value_or_default('ridesharing_greenlet_pool_size', instance_db, self.name)502    @property503    def greenlet_pool_for_ridesharing_services(self):504        # type: () -> bool505        instance_db = self.get_models()506        return get_value_or_default('greenlet_pool_for_ridesharing_services', instance_db, self.name)507    @property508    def max_nb_crowfly_by_mode(self):509        # type: () -> Dict[Text, int]510        instance_db = self.get_models()511        # the value by default is a dict...512        d = copy.deepcopy(get_value_or_default('max_nb_crowfly_by_mode', instance_db, self.name))513        # In case we add a new max_nb_crowfly for an other mode than514        # the ones already present in the database.515        for mode, duration in default_values.max_nb_crowfly_by_mode.items():516            if mode not in d:517                d[mode] = duration518        return d519    @property520    def poi_dataset(self):521        # type: () -> Text522        instance_db = self.get_models()523        return instance_db.poi_dataset if instance_db else None524    @property525    def max_car_no_park_direct_path_duration(self):526        # type: () -> int527        instance_db = self.get_models()528        return get_value_or_default('max_car_no_park_direct_path_duration', instance_db, self.name)529    @property530    def min_ridesharing(self):531        # type: () -> int532        instance_db = self.get_models()533        return get_value_or_default('min_ridesharing', instance_db, self.name)534    @property535    def max_bike_direct_path_distance(self):536        # type: () -> int537        instance_db = self.get_models()538        return get_value_or_default('max_bike_direct_path_distance', instance_db, self.name)539    @property540    def max_bss_direct_path_distance(self):541        # type: () -> int542        instance_db = self.get_models()543        return get_value_or_default('max_bss_direct_path_distance', instance_db, self.name)544    @property545    def max_walking_direct_path_distance(self):546        # type: () -> int547        instance_db = self.get_models()548        return get_value_or_default('max_walking_direct_path_distance', instance_db, self.name)549    @property550    def max_car_direct_path_distance(self):551        # type: () -> int552        instance_db = self.get_models()553        return get_value_or_default('max_car_direct_path_distance', instance_db, self.name)554    @property555    def max_car_no_park_direct_path_distance(self):556        # type: () -> int557        instance_db = self.get_models()558        return get_value_or_default('max_car_no_park_direct_path_distance', instance_db, self.name)559    @property560    def max_ridesharing_direct_path_distance(self):561        # type: () -> int562        instance_db = self.get_models()563        return get_value_or_default('max_ridesharing_direct_path_distance', instance_db, self.name)564    @property565    def max_taxi_direct_path_distance(self):566        # type: () -> int567        instance_db = self.get_models()568        return get_value_or_default('max_taxi_direct_path_distance', instance_db, self.name)569    # TODO: refactorise all properties570    taxi_speed = _make_property_getter('taxi_speed')571    additional_time_after_first_section_taxi = _make_property_getter('additional_time_after_first_section_taxi')572    additional_time_before_last_section_taxi = _make_property_getter('additional_time_before_last_section_taxi')573    max_walking_direct_path_duration = _make_property_getter('max_walking_direct_path_duration')574    max_bike_direct_path_duration = _make_property_getter('max_bike_direct_path_duration')575    max_bss_direct_path_duration = _make_property_getter('max_bss_direct_path_duration')576    max_car_direct_path_duration = _make_property_getter('max_car_direct_path_duration')577    max_taxi_direct_path_duration = _make_property_getter('max_taxi_direct_path_duration')578    max_ridesharing_direct_path_duration = _make_property_getter('max_ridesharing_direct_path_duration')579    street_network_car = _make_property_getter('street_network_car')580    street_network_car_no_park = _make_property_getter('street_network_car_no_park')581    street_network_walking = _make_property_getter('street_network_walking')582    street_network_bike = _make_property_getter('street_network_bike')583    street_network_bss = _make_property_getter('street_network_bss')584    street_network_ridesharing = _make_property_getter('street_network_ridesharing')585    street_network_taxi = _make_property_getter('street_network_taxi')586    stop_points_nearby_duration = _make_property_getter('stop_points_nearby_duration')587    bss_rent_duration = _make_property_getter('bss_rent_duration')588    bss_rent_penalty = _make_property_getter('bss_rent_penalty')589    bss_return_duration = _make_property_getter('bss_return_duration')590    bss_return_penalty = _make_property_getter('bss_rent_penalty')591    asgard_language = _make_property_getter('asgard_language')592    transfer_path = _make_property_getter('transfer_path')593    access_points = _make_property_getter('access_points')594    default_pt_planner = _make_property_getter('default_pt_planner')595    pt_planners_configurations = _make_property_getter('pt_planners_configurations')596    filter_odt_journeys = _make_property_getter('filter_odt_journeys')597    def get_pt_planner(self, pt_planner_id=None):598        pt_planner_id = pt_planner_id or self.default_pt_planner599        return self._pt_planner_manager.get_pt_planner(pt_planner_id)600    @property601    def places_proximity_radius(self):602        # type: () -> int603        instance_db = self.get_models()604        return get_value_or_default('places_proximity_radius', instance_db, self.name)605    def reap_socket(self, ttl):606        # type: (int) -> None607        if self.zmq_socket_type != 'transient':608            return609        logger = logging.getLogger(__name__)610        now = time.time()611        def _reap_sockets(connector):612            while True:613                try:614                    socket, t = connector._sockets.popleft()615                    if now - t > ttl:616                        logger.debug("closing one socket for %s", connector.name)617                        socket.setsockopt(zmq.LINGER, 0)...

command.py

Source:command.py

...40        Model(get_cars(3), 'BMW G450X'),41        Model(get_cars(3), 'BMW R1200GR')]42    db_session.add_all(models)43    db_session.commit()44def get_models(id):45    return Model.query.get(id)46def populate_version():47    versions = [Version(get_models(1), 'DB5 Vantage'),48        Version(get_models(1), 'DB5 convertible'),49        Version(get_models(1), 'DB5 shooting-brake'),50        Version(get_models(2), 'Rapide S 2013'),51        Version(get_models(2), 'RapideE'),52        Version(get_models(2), 'Rapide AMR'),53        Version(get_models(3), 'Series 1 1974-1975'),54        Version(get_models(3), 'Series 2 1976-1985'),55        Version(get_models(6), 'Typ 86'),56        Version(get_models(12), '2008-2016')]57        58    db_session.add_all(versions)59    db_session.commit()60# populate database61@click.command('pop-db')62@with_appcontext63def populate_db_command():64    populate_car()65    populate_model()66    populate_version()67    click.echo('Populate database.')68# 69def init_app(app):70    app.cli.add_command(init_db_command)...

test_cars.py

Source:test_cars.py

...5    assert most_prolific_automaker(2008) == 'Toyota'6def test_most_prolific_automaker_2013():7    assert most_prolific_automaker(2013) == 'Hyundai'8def test_get_models_volkswagen():9    models = get_models('Volkswagen', 2008)10    # sets are unordered11    assert len(models) == 212    assert 'Jetta' in models13    assert 'Rabbit' in models14def test_get_models_nissan():15    assert get_models('Nissan', 2000) == {'Pathfinder'}16def test_get_models_open():17    # not in data set18    assert get_models('Opel', 2008) == set()19def test_get_models_mercedes():20    models = get_models('Mercedes-Benz', 2007)21    assert len(models) == 322    assert 'SL-Class' in models23    assert 'GL-Class' in models...

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.