azure_functions_worker/utils/dependency.py [245:352]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - @classmethod def _add_to_sys_path(cls, path: str, add_to_first: bool): """This will ensure no duplicated path are added into sys.path and clear importer cache. No action if path already exists in sys.path. Parameters ---------- path: str The path needs to be added into sys.path. If the path is an empty string, no action will be taken. add_to_first: bool Should the path added to the first entry (highest priority) """ if path and path not in sys.path: if add_to_first: sys.path.insert(0, path) else: sys.path.append(path) # Only clear path importer and sys.modules cache if path is not # defined in sys.path cls._clear_path_importer_cache_and_modules(path) @classmethod def _remove_from_sys_path(cls, path: str): """This will remove path from sys.path and clear importer cache. No action if the path does not exist in sys.path. Parameters ---------- path: str The path to be removed from sys.path. If the path is an empty string, no action will be taken. """ if path and path in sys.path: # Remove all occurances in sys.path sys.path = list(filter(lambda p: p != path, sys.path)) # In case if any part of worker initialization do sys.path.pop() # Always do a cache clear in path importer and sys.modules cls._clear_path_importer_cache_and_modules(path) @classmethod def _clear_path_importer_cache_and_modules(cls, path: str): """Removes path from sys.path_importer_cache and clear related sys.modules cache. No action if the path is empty or no entries in sys.path_importer_cache or sys.modules. Parameters ---------- path: str The path to be removed from sys.path_importer_cache. All related modules will be cleared out from sys.modules cache. If the path is an empty string, no action will be taken. """ if path and path in sys.path_importer_cache: sys.path_importer_cache.pop(path) if path: cls._remove_module_cache(path) @staticmethod def _get_cx_deps_path() -> str: """Get the directory storing the customer's third-party libraries. Returns ------- str Core Tools: path to customer's site pacakges Linux Dedicated/Premium: path to customer's site pacakges Linux Consumption: empty string """ prefix: Optional[str] = os.getenv(AZURE_WEBJOBS_SCRIPT_ROOT) cx_paths: List[str] = [ p for p in sys.path if prefix and p.startswith(prefix) and ('site-packages' in p) ] # Return first or default of customer path return (cx_paths or [''])[0] @staticmethod def _get_cx_working_dir() -> str: """Get the customer's working directory. Returns ------- str Core Tools: AzureWebJobsScriptRoot env variable Linux Dedicated/Premium: AzureWebJobsScriptRoot env variable Linux Consumption: empty string """ return os.getenv(AZURE_WEBJOBS_SCRIPT_ROOT, '') @staticmethod def _get_worker_deps_path() -> str: """Get the worker dependency sys.path. This will always available even in all skus. Returns ------- str The worker packages path """ # 1. Try to parse the absolute path python/3.8/LINUX/X64 in sys.path r = re.compile(r'.*python(\/|\\)\d+\.\d+(\/|\\)(WINDOWS|LINUX|OSX).*') worker_deps_paths: List[str] = [p for p in sys.path if r.match(p)] if worker_deps_paths: return worker_deps_paths[0] - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - proxy_worker/utils/dependency.py [153:260]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - @classmethod def _add_to_sys_path(cls, path: str, add_to_first: bool): """This will ensure no duplicated path are added into sys.path and clear importer cache. No action if path already exists in sys.path. Parameters ---------- path: str The path needs to be added into sys.path. If the path is an empty string, no action will be taken. add_to_first: bool Should the path added to the first entry (highest priority) """ if path and path not in sys.path: if add_to_first: sys.path.insert(0, path) else: sys.path.append(path) # Only clear path importer and sys.modules cache if path is not # defined in sys.path cls._clear_path_importer_cache_and_modules(path) @classmethod def _remove_from_sys_path(cls, path: str): """This will remove path from sys.path and clear importer cache. No action if the path does not exist in sys.path. Parameters ---------- path: str The path to be removed from sys.path. If the path is an empty string, no action will be taken. """ if path and path in sys.path: # Remove all occurances in sys.path sys.path = list(filter(lambda p: p != path, sys.path)) # In case if any part of worker initialization do sys.path.pop() # Always do a cache clear in path importer and sys.modules cls._clear_path_importer_cache_and_modules(path) @classmethod def _clear_path_importer_cache_and_modules(cls, path: str): """Removes path from sys.path_importer_cache and clear related sys.modules cache. No action if the path is empty or no entries in sys.path_importer_cache or sys.modules. Parameters ---------- path: str The path to be removed from sys.path_importer_cache. All related modules will be cleared out from sys.modules cache. If the path is an empty string, no action will be taken. """ if path and path in sys.path_importer_cache: sys.path_importer_cache.pop(path) if path: cls._remove_module_cache(path) @staticmethod def _get_cx_deps_path() -> str: """Get the directory storing the customer's third-party libraries. Returns ------- str Core Tools: path to customer's site packages Linux Dedicated/Premium: path to customer's site packages Linux Consumption: empty string """ prefix: Optional[str] = os.getenv(AZURE_WEBJOBS_SCRIPT_ROOT) cx_paths: List[str] = [ p for p in sys.path if prefix and p.startswith(prefix) and ('site-packages' in p) ] # Return first or default of customer path return (cx_paths or [''])[0] @staticmethod def _get_cx_working_dir() -> str: """Get the customer's working directory. Returns ------- str Core Tools: AzureWebJobsScriptRoot env variable Linux Dedicated/Premium: AzureWebJobsScriptRoot env variable Linux Consumption: empty string """ return os.getenv(AZURE_WEBJOBS_SCRIPT_ROOT, '') @staticmethod def _get_worker_deps_path() -> str: """Get the worker dependency sys.path. This will always available even in all skus. Returns ------- str The worker packages path """ # 1. Try to parse the absolute path python/3.13/LINUX/X64 in sys.path r = re.compile(r'.*python(\/|\\)\d+\.\d+(\/|\\)(WINDOWS|LINUX|OSX).*') worker_deps_paths: List[str] = [p for p in sys.path if r.match(p)] if worker_deps_paths: return worker_deps_paths[0] - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -