KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use get_environment_variables method in localstack
Best Python code snippet using localstack_python
Project.py
Source:Project.py 
...42 def get_snakefile_directory(self) -> Path:43 raise Exception("Should not be called!")44 def get_main_snakefile_path(self) -> Path:45 raise Exception("Should not be called!")46 def get_environment_variables(self) -> Dict[str, List[str]]:47 raise Exception("Should not be called!")48 def get_definition(self) -> str:49 '''50 Definitions is a string of the form:51 "directory:<dir>" OR52 "package:<package>" OR53 "<dir>" which is equivallent to "directory:<dir>"54 '''55 raise Exception("Should not be called!")56 def build_object_graph(self, project: Project, args: List[str]) \57 -> ObjectGraph:58 project = project59 args = args60 raise Exception("not yet")61class DirectoryPipeline(Pipeline):62 """63 Subclass of the Pipeline class defined by directory path.64 A snakeobject DirectoryPipeline attributes are:65 .. py:attribute:: definition_dir66 :type: str67 .. py:attribute:: snakefile_directory68 :type: str69 DirecoryPipeline useful functions are:70 .. py:attribute:: get_snakefile_directory()71 returns directory of Snakefile72 .. py:attribute:: get_main_snakefile_path()73 returns path to the Snakefile74 .. py:attribute:: get_environment_variables()75 returns dictionary with keys PATH, PYTHPONPATH, etc.76 .. py:attribute:: build_object_graph(proj: Project, args: List[str])77 returns instance of and ObjectGraph78 """79 def __init__(self, definition_dir: str):80 self.definition_dir = definition_dir81 self.snakefile_directory = Path(definition_dir).resolve(True)82 def get_snakefile_directory(self) -> Path:83 return self.snakefile_directory84 def get_main_snakefile_path(self) -> Path:85 return self.snakefile_directory / "Snakefile"86 DIR_TO_ENV = {87 "bin": ["PATH"],88 "python": ["PATH", "PYTHONPATH"],89 "R": ["PATH"],90 "perl": ["PATH", "PERL5LIB"]91 }92 def get_environment_variables(self) -> Dict[str, List[str]]:93 vars: Dict[str, List[str]] = {}94 vars["PATH"] = [str(self.snakefile_directory)]95 for sub_dir, env_vars in DirectoryPipeline.DIR_TO_ENV.items():96 sub_dir_path = self.snakefile_directory / sub_dir97 if os.path.exists(sub_dir_path):98 for var in env_vars:99 try:100 vars[var].append(str(sub_dir_path))101 except KeyError:102 vars[var] = [str(sub_dir_path)]103 return vars104 def get_definition(self) -> str:105 return self.definition_dir106 def build_object_graph(self, project: Project, args: List[str]) \107 -> ObjectGraph:108 bldObjGraphPy = self.get_snakefile_directory() / "build_object_graph.py"109 if not os.path.isfile(bldObjGraphPy):110 raise Exception(f'ERROR: There is no {bldObjGraphPy}')111 spec = spec_from_file_location("build_object_graph", bldObjGraphPy)112 if spec is None:113 raise Exception("Could build specs for build_object_graph")114 if spec.loader is None:115 raise Exception("Loader is empty??")116 foo = module_from_spec(spec)117 spec.loader.exec_module(foo)118 newObjectGraph = ObjectGraph()119 foo.run(project, newObjectGraph, *args)120 return newObjectGraph121class PackagePipeline(Pipeline):122 """123 Subclass of Pipeline class defined by python package.124 A snakeobject PackagePipeline attributes are:125 .. py:attribute:: definition_package126 :type: str127 .. py:attribute:: snake_file_dir128 :type: str129 PackagePipeline useful functions are:130 .. py:attribute:: get_snakefile_directory()131 returns directory of Snakefile132 .. py:attribute:: get_main_snakefile_path()133 returns path to the Snakefile134 .. py:attribute:: get_environment_variables()135 returns dictionary with keys PATH, PYTHPONPATH, etc.136 .. py:attribute:: build_object_graph(proj: Project, args: List[str])137 returns instance of and ObjectGraph138 """139 def __init__(self, definition_package: str):140 self.definition_package = definition_package141 self.pipeline_package = importlib.import_module(self.definition_package)142 if self.pipeline_package.__file__ is None:143 raise Exception(f"{definition_package} should be a python package 1")144 if not self.pipeline_package.__file__.endswith('__init__.py'):145 raise Exception(f"{definition_package} should be a python package 2")146 self.snake_file_dir = Path(self.pipeline_package.__file__).parent147 def get_snakefile_directory(self) -> Path:148 return self.snake_file_dir149 def get_main_snakefile_path(self) -> Path:150 return self.snake_file_dir / "Snakefile"151 def get_environment_variables(self) -> Dict[str, List[str]]:152 try:153 return self.pipeline_package.get_environment_variables()154 except:155 return {}156 157 def get_definition(self) -> str:158 return f"pacakage:{self.definition_package}"159 def build_object_graph(self, project: Project, args: List[str]) \160 -> ObjectGraph:161 new_object_graph = ObjectGraph()162 self.pipeline_package.build_object_graph(163 project, new_object_graph, args)164 return new_object_graph165def build_pipeline(pipeline_definition: str) -> Pipeline:166 if pipeline_definition.startswith('directory:'):167 dir_path = pipeline_definition.split(':')[1]168 return DirectoryPipeline(dir_path)169 if pipeline_definition.startswith('package'):170 package_name = pipeline_definition.split(':')[1]171 return PackagePipeline(package_name)172 return DirectoryPipeline(pipeline_definition)173class Project:174 """175 Each objects of the Project class represents a snakeobject project.176 The project directory is given as the ``directory`` parameter. If 177 ``directory`` is none,178 the :py:func:`find_so_project_directory` function is used to determine 179 the project directory.180 A snakeobject project attributes are:181 .. py:attribute:: directory182 :type: str183 the project directory184 .. py:attribute:: parameters185 :type: dict[str,str]186 a key value dictionary for global project level parameters187 .. py:attribute:: parent_projects188 :type: dict[str,Project]189 a key value dictionary for parent project names190 .. py:attribute:: objectGraph191 :type: ObjectGraph192 the objectGraph for the project193 Project useful functions are:194 .. py:attribute:: get_object_directory(object_instance)195 returns directory of object196 .. py:attribute:: get_object_path(type_name, object_name)197 returns path to the object of type `type_name`, and object name `object_name`198 .. py:attribute:: get_pipeline_directory( )199 returns path to the project pipeline directory200 .. py:attribute:: get_parameter(name, parent_project_id=None)201 If `parent_project_id` is given returns the value of parameter in202 that project, otherwise, if `name` is in project parameters, returns203 its value. Finally, if `name` is not in the project parameters,204 returns the value of named parameters in the first project205 recursively folowing parent projects.206 """207 def __init__(self, directory=None):208 self.directory = directory if directory else find_so_project_directory()209 self._porojectFileName = self.directory + "/so_project.yaml"210 self._objectGraphFileName = self.directory + "/OG.json"211 if os.path.isfile(self._porojectFileName):212 self.parameters = load_yaml(self._porojectFileName)213 self.parameters['directory'] = self.directory214 else:215 self.parameters = {}216 self.parent_projects: Dict[str, Project] = {}217 self._run_parameter_interpolation()218 if 'so_parent_projects' in self.parameters and not self.parent_projects:219 for p, d in self.parameters['so_parent_projects'].items():220 self.parent_projects[p] = Project(d)221 if os.path.isfile(self._objectGraphFileName):222 self.objectGraph = load_object_graph(self._objectGraphFileName)223 else:224 self.objectGraph = ObjectGraph()225 self.set_pipeline()226 def interpolate(self, obj, oo=None):227 ptn = re.compile(r'(\[(\w+)\:([\w\/]+)(\:(\w+))?\])(\w*)')228 if type(obj) == int or type(obj) == float:229 return obj230 elif type(obj) == list:231 return [self.interpolate(x) for x in obj]232 elif type(obj) == dict:233 return {k: self.interpolate(v) for k, v in obj.items()}234 elif type(obj) == str:235 for s in ptn.findall(obj):236 iType = s[1]237 name = s[2]238 prnt = s[4]239 # print("AAA",O,s)240 if iType == 'P':241 if not oo:242 raise ProjectException("object interpolation requires an object")243 if name not in oo.params:244 raise ProjectException(245 "parameter %s is not defined for the object %s" % (name, oo.k()))246 obj = obj.replace(s[0], oo.params[name])247 elif iType == 'E':248 if name not in os.environ:249 raise ProjectException("Varianble %s is not defined" % name)250 obj = obj.replace(s[0], os.environ[name])251 elif iType == 'D':252 if name == "project":253 pv = self.directory254 elif name == "pipeline":255 pv = str(self.get_pipeline_directory())256 else:257 raise ProjectException('The project property %s is unknonw.' % name)258 obj = obj.replace(s[0], pv)259 elif iType == 'PP':260 if ('so_parent_projects' in self.parameters and261 not self.parent_projects):262 for k, v in self.parameters['so_parent_projects'].items():263 self.parent_projects[k] = Project(v)264 if not prnt:265 if name not in self.parameters:266 raise ProjectException('Parameter %s is not defined' % name)267 obj = obj.replace(s[0], self.parameters[name])268 elif '/' not in prnt:269 if '/' not in prnt:270 if prnt in self.parent_projects:271 dir = self.parent_projects[prnt].directory272 if not os.path.exists(dir):273 raise ProjectException('The path to parent does not exist')274 self.parent_projects[prnt] = Project(dir)275 obj = obj.replace(276 s[0], str(self.parent_projects[prnt].parameters[name]))277 else:278 raise ProjectException(279 f'Project {prnt} is not in so_parent_projects')280 elif '/' in prnt:281 cs = prnt.split('/')282 if not cs[0] in self.parent_projects:283 raise ProjectException(f'Project {cs[0]} is not in so_parent_projects')284 else:285 dir = self.parent_projects[cs[0]]286 if not os.path.exists(dir):287 raise ProjectException(f'The path {dir} to parent does not exist')288 self.parent_projects[cs[0]] = Project(dir)289 P = self.parent_projects[cs[0]]290 for p in cs[1:]:291 P = P.parent_projects[p]292 obj = obj.replace(s[0], str(P.parameters[name]))293 else:294 raise ProjectException(295 'Interpolation type [%s: ...] is unknown; can be only E|P|PP|D.' % iType)296 return obj297 def _run_parameter_interpolation(self):298 for k, v in sorted(self.parameters.items(), key=lambda x: not '[D:' in x):299 self.parameters[k] = self.interpolate(v)300 def get_parent_project(self, parent_project_id):301 parent_project_ids = parent_project_id.split("/")302 proj = self303 for id in parent_project_ids:304 proj = proj.parent_projects[id]305 return proj306 def get_parameter(self, name, parent_project_id=None):307 if parent_project_id:308 proj = self.get_parent_project(parent_project_id)309 return proj.parameters[name]310 if name in self.parameters:311 return self.parameters[name]312 for proj in self.parent_projects.values():313 v = proj.get_parameter(name)314 if v:315 return v316 return None317 def set_pipeline(self):318 pipeline_definition = f"directory:{self.directory}"319 if "SO_PIPELINE" in os.environ:320 pipeline_definition = os.environ['SO_PIPELINE']321 if "so_pipeline" in self.parameters:322 pipeline_definition = self.parameters["so_pipeline"]323 self.pipeline = build_pipeline(pipeline_definition)324 '''325 if "so_pipeline" in self.parameters:326 ppd = self.parameters['so_pipeline']327 if ppd.startswith('directory:'):328 ppd = os.path.abspath(ppd.split(':')[1])329 self.pipeline = DirectoryPipeline(ppd)330 elif ppd.startswith('package'):331 self.pipeline = PackagePipeline(ppd.split(':')[1])332 else:333 self.pipeline = DirectoryPipeline(ppd)334 elif "SO_PIPELINE" in os.environ:335 self.pipeline = DirectoryPipeline(os.environ['SO_PIPELINE'])336 elif os.path.exists("workflow"):337 self.pipeline = DirectoryPipeline(os.path.abspath("workflow"))338 else:339 self.pipeline = DirectoryPipeline(os.path.abspath(self.directory))340 '''341 def get_pipeline_directory(self) -> Path:342 self.set_pipeline()343 return self.pipeline.get_snakefile_directory()344 def get_environment_variables(self) -> Dict[str, List[str]]:345 vars = defaultdict(list)346 def add_vars(extra_vars: Dict[str, List[str]]):347 for var, values in extra_vars.items():348 try:349 vars[var] += values350 except KeyError:351 vars[var] = values352 # so_environment_EEE_add353 add_env_re = re.compile("so_environment_(.*)_add")354 for param_key, param_value in self.parameters.items():355 match = add_env_re.match(param_key)356 if match:357 env_name = match.group(1)358 if type(param_value) == list:359 for pv in param_value:360 vars[env_name].append(pv)361 else:362 vars[env_name].append(param_value)363 add_vars(self.pipeline.get_environment_variables())364 for pp in self.parent_projects.values():365 add_vars(pp.get_environment_variables())366 # so_environment_EEE_set367 set_env_re = re.compile("so_environment_(.*)_set")368 for param_key, param_value in self.parameters.items():369 match = set_env_re.match(param_key)370 if match:371 env_name = match.group(1)372 vars[env_name] = [param_value]373 return vars374 def save_object_graph(self):375 self.objectGraph.save(self._objectGraphFileName)376 def ensure_object_type_snakefile_exists(self, ot):377 sfile = self.get_pipeline_directory() / (ot + ".snakefile")378 if not os.path.exists(sfile):379 print(f'WARNING: creating dummy snakefile '380 f'{str(sfile).split("/")[-1]}')381 with open(sfile, 'w') as f:382 f.write(f'add_targets()\n')383 return sfile384 def write_main_snakefile(self):385 from glob import glob386 mf = self.get_pipeline_directory() / "Snakefile"387 from snakeobjects import __version__388 o_types = set(self.objectGraph.get_object_types())389 o_list = set([x.split('/')[-1][:-10] for x in390 glob(str(self.get_pipeline_directory() / '*.snakefile'))])391 all_o_types = sorted(set.union(o_types, o_list))392 if os.path.exists(mf):393 with open(mf) as f:394 old_version = f.readline().strip('\n\r').split(' ')[1]395 old_o_types = f.readline().strip('\n\r').split(' ')[2]396 if old_version == __version__ and old_o_types == ','.join(all_o_types):397 return398 header = importlib_resources.read_text(__package__, 'header.snakefile')399 with open(mf, 'w') as f:400 f.write(f'#snakeobjects {__version__}\n')401 f.write(f'#Snakeobjects types: {all_o_types}\n')402 f.write(header)403 for ot in all_o_types:404 sfile = self.ensure_object_type_snakefile_exists(ot)405 sfile = ot + ".snakefile"406 f.write(f'include: "{sfile}"\n')407 f.write(f'rule so_all_{ot}:\n')408 f.write(f' input:\n')409 f.write(f' expand("{{of}}", of=project.get_all_object_flags("{ot}"))\n\n')410 f.write(f'rule so_{ot}_obj:\n')411 f.write(f' input: get_targets("{ot}")\n')412 f.write(f' output: touch("{ot}/{{oid}}/obj.flag")\n\n')413 def get_object_flag(self, o):414 return f'{o.oType}/{o.oId}/obj.flag'415 def get_object_directory(self, o):416 return f'{self.directory}/{o.oType}/{o.oId}'417 def get_object_path(self, oType, oId):418 return Path(self.get_object_directory(self.objectGraph[oType, oId]))419 def create_symbolic_links(self):420 for tp in sorted(self.objectGraph.get_object_types()):421 for o in self.objectGraph[tp]:422 oDir = self.get_object_directory(o)423 for k, v in list(o.params.items()):424 if not k.startswith("symlink."):425 continue426 if not os.path.exists(oDir):427 os.makedirs(oDir)428 dst = oDir + "/" + k[8:]429 src = v430 os.system("ln -sf %s %s" % (src, dst))431 os.system("touch -h -r %s %s" % (src, dst))432 def get_all_object_flags(self, oType=None):433 OG = self.objectGraph434 if oType:435 return [self.get_object_flag(o) for o in OG[oType]]436 return [self.get_object_flag(o) for oType in OG.get_object_types() for o in OG[oType]]437 def set_environment(self,update_environment=True):438 if update_environment:439 print("UPDATING ENVIRONMENT:")440 print("export SO_PROJECT=", self.directory, sep="")441 print("export SO_PIPELINE=", self.pipeline.get_definition(), sep="")442 os.environ['SO_PROJECT'] = self.directory443 os.environ['SO_PIPELINE'] = str(self.pipeline.get_definition())444 vars = self.get_environment_variables()445 for var, values in vars.items():446 my_val = os.pathsep.join(values)447 if var in ['PATH', 'PYTHONPATH', 'PERL5LIB'] and var in os.environ:448 print(f"export {var}={my_val}{os.pathsep}${var}")449 values.append(os.environ[var])450 else:451 print(f"export {var}={my_val}")452 full_val = os.pathsep.join(values)453 os.environ[var] = full_val454 def buildObjectGraph(self, args: List[str]):455 self.set_environment()456 self.objectGraph = self.pipeline.build_object_graph(self, args)457if __name__ == "__main__":458 print("hi")...
command_handler.py
Source:command_handler.py
...60 runner = DockerPlanRunner(project_compose=self.project_compose,61 working_directory=self.working_directory,62 repo_dir=self.repo_dir)63 if StateHolder.always_update and cmd in ('start', 'up', 'restart'): # Pull before start in developer mode64 runner.run(plan=plan, commands='pull', envs=self.get_environment_variables(plan=plan))65 for cmd in command_list['docker']:66 if cmd == 'pull' and StateHolder.offline: # Skip pull in offline mode67 continue68 runner.run(plan=plan, commands=cmd,69 envs=self.get_environment_variables(plan=plan))70 def run_kubernetes(self, cmd, command_list, plan):71 runner = KubernetesRunner(working_directory=self.working_directory, repo_dir=self.repo_dir) \72 if StateHolder.container_mode == 'Kubernetes' \73 else HelmRunner(working_directory=self.working_directory, repo_dir=self.repo_dir)74 if len(command_list[StateHolder.container_mode.lower()]) == 0:75 ColorPrint.exit_after_print_messages('Command: ' + cmd + ' not supported with' + StateHolder.container_mode)76 for cmd in command_list[StateHolder.container_mode.lower()]:77 runner.run(plan=plan, commands=cmd, envs=self.get_environment_variables(plan=plan))78 def check_command(self, cmd):79 if self.hierarchy is None or not isinstance(self.hierarchy, dict):80 ColorPrint.exit_after_print_messages("Command hierarchy config is missing")81 if cmd not in self.hierarchy:82 ColorPrint.exit_after_print_messages("Command not found in hierarchy: " + str(cmd))83 def pre_run(self, command_list, plan):84 if command_list.get('before', False):85 self.script_runner.run(plan=plan, script_type='before_script')86 self.run_method('premethods', command_list)87 def after_run(self, command_list, plan):88 self.run_method('postmethods', command_list)89 if command_list.get('after', False):90 self.script_runner.run(plan=plan, script_type='after_script')91 def run_method(self, type, command_list):92 if type in command_list and len(command_list[type]) > 0:93 for method in command_list[type]:94 getattr(self, method)()95 def parse_environment_dict(self, path, env):96 """Compose dictionary from environment variables."""97 file_name = FileUtils.get_compose_file_relative_path(repo_dir=self.repo_dir,98 working_directory=self.working_directory,99 file_name=path)100 env_file = ProjectUtils.get_file(file=file_name)101 if env_file is None:102 ColorPrint.exit_after_print_messages(103 message="Environment file (" + str(file_name) + ") not exists in repository: " + StateHolder.name)104 with open(env_file) as stream:105 for line in stream.readlines():106 if not line.startswith("#"):107 data = line.split("=", 1)108 if len(data) > 1:109 env[data[0].strip()] = data[1].strip()110 def get_environment_dict(self, envs):111 """Process environment files. Environment for selected plan will be override the defaults"""112 environment = dict()113 '''First the default, if exists'''114 if "environment" in self.project_compose:115 self.parse_environment_dict(path=self.project_compose["environment"]["include"], env=environment)116 for env in envs:117 self.parse_environment_dict(path=env, env=environment)118 return environment119 def get_environment_variables(self, plan):120 """Get all environment variables depends on selected plan"""121 envs = list()122 if isinstance(plan, dict):123 if 'environment' in plan and 'include' in plan['environment']:124 envs.extend(ProjectUtils.get_list_value(plan['environment']['include']))125 if 'docker-compose-dir' in plan:126 envs.extend(FileUtils.get_filtered_sorted_alter_from_base_dir(base_dir=self.repo_dir,127 actual_dir=self.working_directory,128 target_directories=ProjectUtils.129 get_list_value(130 plan['docker-compose-dir']),131 filter_ends='.env'))132 env_dict = self.get_environment_dict(envs=envs)133 env_copy = os.environ.copy()134 for key in env_dict.keys():135 env_copy[key] = env_dict[key]136 """Add host system to environment"""137 env_copy["HOST_SYSTEM"] = platform.system()138 return env_copy139 def pack(self):140 plan = self.project_compose['plan'][self.plan]141 envs = self.get_environment_variables(plan=plan)142 runner = DockerPlanRunner(project_compose=self.project_compose,143 working_directory=self.working_directory,144 repo_dir=self.repo_dir)...
config_pipeline.py
Source:config_pipeline.py
1"""2This is a boilerplate pipeline 'example'3generated using Kedro 0.18.04"""5from kedro.pipeline import Pipeline, node, pipeline6from .nodes import get_environment_variables7def create_pipeline(**kwargs) -> Pipeline:8 # TODO: cover by test case9 return pipeline(10 [11 node(12 func=get_environment_variables,13 inputs=None,14 outputs="02_intermediate__env_variables",15 name="config__env_vars_extraction",16 ),17 node(18 func=lambda x: x["task_id"],19 inputs="02_intermediate__env_variables",20 outputs="params:common.task_id",21 name="geocoding__config__extraction__task_id",22 ),23 ],...
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?
