KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
