src/use/use.py
def _use_path(
        self,
        path,
        /,
        *,
        initial_globals=None,
        as_import: str = None,
        default=mode.fastfail,
        modes=0,
    ) -> ProxyModule:
        """Import a module from a path.

        https://github.com/amogorkon/justuse/wiki/Use-Path

        Args:
            path ([type]): must be a pathlib.Path
            initial_globals ([type], optional): dict that should be globally available to the module before executing it. Defaults to None.
            default ([type], optional): Return instead if an exception is encountered.
            modes (int, optional): [description]. Defaults to 0; Acceptable mode for this variant: use.reloading.

        Returns:
            Optional[ModuleType]: The module if it was imported, otherwise whatever was specified as default.
        """
        log.debug(f"use-path: {path}")
        initial_globals = initial_globals or {}

        reloading = bool(Use.reloading & modes)

        exc = None
        mod = None

        if path.is_dir():
            return _fail_or_default(
                ImportError(f"Can't import directory {path}"), default
            )

        original_cwd = source_dir = Path.cwd()
        try:
            if not path.is_absolute():
                source_dir = Path.cwd()

            # calling from another use()d module
            # let's see where we started
            main_mod = __import__("__main__")
            if source_dir and source_dir.exists():
                os.chdir(source_dir.parent)
                source_dir = source_dir.parent
            else:
                # there are a number of ways to call use() from a non-use() starting point
                # let's first check if we are running in jupyter
                jupyter = "ipykernel" in sys.modules
                # we're in jupyter, we use the CWD as set in the notebook
                if not jupyter and hasattr(main_mod, "__file__"):
                    source_dir = (
                        _ensure_path(
                            inspect.currentframe().f_back.f_back.f_code.co_filename
                        )
                        .resolve()
                        .parent
                    )
            if source_dir is None:
                if main_mod.__loader__ and hasattr(main_mod.__loader__, "path"):
                    source_dir = _ensure_path(main_mod.__loader__.path).parent
                else:
                    source_dir = Path.cwd()
            if not source_dir.joinpath(path).exists():
                if files := [*[*source_dir.rglob(f"**/{path}")]]:
                    source_dir = _ensure_path(files[0]).parent
                else:
                    source_dir = Path.cwd()
            if not source_dir.exists():
                return _fail_or_default(
                    NotImplementedError(
                        "Can't determine a relative path from a virtual file."
                    ),
                    default,
                )
            path = source_dir.joinpath(path).resolve()
            if not path.exists():
                return _fail_or_default(ImportError(f"Sure '{path}' exists?"), default)
            os.chdir(path.parent)
            name = path.stem
            if reloading:
                try:
                    with open(path, "rb") as rfile:
                        code = rfile.read()
                    # initial instance, if this doesn't work, just throw the towel
                    mod = _build_mod(
                        name=name,
                        code=code,
                        initial_globals=initial_globals,
                        module_path=path.resolve(),
                    )
                except KeyError:
                    exc = traceback.format_exc()
                if exc:
                    return _fail_or_default(ImportError(exc), default)
                mod = ProxyModule(mod)
                reloader = ModuleReloader(
                    proxy=mod,
                    name=name,
                    path=path,
                    initial_globals=initial_globals,
                )
                _reloaders[mod] = reloader

                threaded = False
                # this looks like a hack, but isn't one -
                # jupyter is running an async loop internally, which works better async than threaded!
                try:
                    asyncio.get_running_loop()
                # we're dealing with non-async code, we need threading
                except RuntimeError:
                    # can't have the code inside the handler because of "during handling of X, another exception Y happened"
                    threaded = True
                if not threaded:
                    reloader.start_async()
                else:
                    reloader.start_threaded()

                if not all(
                    isfunction(value)
                    for key, value in mod.__dict__.items()
                    if key not in initial_globals.keys() and not key.startswith("__")
                ):
                    warn(Message.not_reloadable(name), NotReloadableWarning)
            else:  # NOT reloading
                with open(path, "rb") as rfile:
                    code = rfile.read()
                # the path needs to be set before attempting to load the new module - recursion confusing ftw!
                frame = inspect.getframeinfo(inspect.currentframe())
                self._set_mod(name=name, mod=mod, frame=frame)
                try:
                    mod = _build_mod(
                        name=name,
                        code=code,
                        initial_globals=initial_globals,
                        module_path=path,
                    )
                except KeyError:
                    del self._using[name]
                    exc = traceback.format_exc()
        except KeyError:
            exc = traceback.format_exc()
        finally:
            # let's not confuse the user and restore the cwd to the original in any case
            os.chdir(original_cwd)
        if exc:
            return _fail_or_default(ImportError(exc), default)
        if as_import:
            sys.modules[as_import] = mod
        frame = inspect.getframeinfo(inspect.currentframe())
        self._set_mod(name=name, mod=mod, frame=frame)
        return ProxyModule(mod)
Import a module from a path.

        https://github.com/amogorkon/justuse/wiki/Use-Path

        Args:
            path ([type]): must be a pathlib.Path
            initial_globals ([type], optional): dict that should be globally available to the module before executing it. Defaults to None.
            default ([type], optional): Return instead if an exception is encountered.
            modes (int, optional): [description]. Defaults to 0; Acceptable mode for this variant: use.reloading.

        Returns:
            Optional[ModuleType]: The module if it was imported, otherwise whatever was specified as default.
        
Import a module from a path.
Import a module from a path .
def _use_path(
        self,
        path,
        /,
        *,
        initial_globals=None,
        as_import: str = None,
        default=mode.fastfail,
        modes=0,
    ) -> ProxyModule:
        log.debug(f"use-path: {path}")
        initial_globals = initial_globals or {}
        reloading = bool(Use.reloading & modes)
        exc = None
        mod = None
        if path.is_dir():
            return _fail_or_default(
                ImportError(f"Can't import directory {path}"), default
            )
        original_cwd = source_dir = Path.cwd()
        try:
            if not path.is_absolute():
                source_dir = Path.cwd()
            main_mod = __import__("__main__")
            if source_dir and source_dir.exists():
                os.chdir(source_dir.parent)
                source_dir = source_dir.parent
            else:
                jupyter = "ipykernel" in sys.modules
                if not jupyter and hasattr(main_mod, "__file__"):
                    source_dir = (
                        _ensure_path(
                            inspect.currentframe().f_back.f_back.f_code.co_filename
                        )
                        .resolve()
                        .parent
                    )
            if source_dir is None:
                if main_mod.__loader__ and hasattr(main_mod.__loader__, "path"):
                    source_dir = _ensure_path(main_mod.__loader__.path).parent
                else:
                    source_dir = Path.cwd()
            if not source_dir.joinpath(path).exists():
                if files := [*[*source_dir.rglob(f"**/{path}")]]:
                    source_dir = _ensure_path(files[0]).parent
                else:
                    source_dir = Path.cwd()
            if not source_dir.exists():
                return _fail_or_default(
                    NotImplementedError(
                        "Can't determine a relative path from a virtual file."
                    ),
                    default,
                )
            path = source_dir.joinpath(path).resolve()
            if not path.exists():
                return _fail_or_default(ImportError(f"Sure '{path}' exists?"), default)
            os.chdir(path.parent)
            name = path.stem
            if reloading:
                try:
                    with open(path, "rb") as rfile:
                        code = rfile.read()
                    mod = _build_mod(
                        name=name,
                        code=code,
                        initial_globals=initial_globals,
                        module_path=path.resolve(),
                    )
                except KeyError:
                    exc = traceback.format_exc()
                if exc:
                    return _fail_or_default(ImportError(exc), default)
                mod = ProxyModule(mod)
                reloader = ModuleReloader(
                    proxy=mod,
                    name=name,
                    path=path,
                    initial_globals=initial_globals,
                )
                _reloaders[mod] = reloader
                threaded = False
                try:
                    asyncio.get_running_loop()
                except RuntimeError:
                    threaded = True
                if not threaded:
                    reloader.start_async()
                else:
                    reloader.start_threaded()
                if not all(
                    isfunction(value)
                    for key, value in mod.__dict__.items()
                    if key not in initial_globals.keys() and not key.startswith("__")
                ):
                    warn(Message.not_reloadable(name), NotReloadableWarning)
            else:  
                with open(path, "rb") as rfile:
                    code = rfile.read()
                frame = inspect.getframeinfo(inspect.currentframe())
                self._set_mod(name=name, mod=mod, frame=frame)
                try:
                    mod = _build_mod(
                        name=name,
                        code=code,
                        initial_globals=initial_globals,
                        module_path=path,
                    )
                except KeyError:
                    del self._using[name]
                    exc = traceback.format_exc()
        except KeyError:
            exc = traceback.format_exc()
        finally:
            os.chdir(original_cwd)
        if exc:
            return _fail_or_default(ImportError(exc), default)
        if as_import:
            sys.modules[as_import] = mod
        frame = inspect.getframeinfo(inspect.currentframe())
        self._set_mod(name=name, mod=mod, frame=frame)
        return ProxyModule(mod)
def _use_path ( self , path , / , * , initial_globals = None , as_import : str = None , default = mode . fastfail , modes = 0 , ) -> ProxyModule : log . debug ( f"use-path: {path}" ) initial_globals = initial_globals or { } reloading = bool ( Use . reloading & modes ) exc = None mod = None if path . is_dir ( ) : return _fail_or_default ( ImportError ( f"Can't import directory {path}" ) , default ) original_cwd = source_dir = Path . cwd ( ) try : if not path . is_absolute ( ) : source_dir = Path . cwd ( ) main_mod = __import__ ( "__main__" ) if source_dir and source_dir . exists ( ) : os . chdir ( source_dir . parent ) source_dir = source_dir . parent else : jupyter = "ipykernel" in sys . modules if not jupyter and hasattr ( main_mod , "__file__" ) : source_dir = ( _ensure_path ( inspect . currentframe ( ) . f_back . f_back . f_code . co_filename ) . resolve ( ) . parent ) if source_dir is None : if main_mod . __loader__ and hasattr ( main_mod . __loader__ , "path" ) : source_dir = _ensure_path ( main_mod . __loader__ . path ) . parent else : source_dir = Path . cwd ( ) if not source_dir . joinpath ( path ) . exists ( ) : if files := [ * [ * source_dir . rglob ( f"**/{path}" ) ] ] : source_dir = _ensure_path ( files [ 0 ] ) . parent else : source_dir = Path . cwd ( ) if not source_dir . exists ( ) : return _fail_or_default ( NotImplementedError ( "Can't determine a relative path from a virtual file." ) , default , ) path = source_dir . joinpath ( path ) . resolve ( ) if not path . exists ( ) : return _fail_or_default ( ImportError ( f"Sure '{path}' exists?" ) , default ) os . chdir ( path . parent ) name = path . stem if reloading : try : with open ( path , "rb" ) as rfile : code = rfile . read ( ) mod = _build_mod ( name = name , code = code , initial_globals = initial_globals , module_path = path . resolve ( ) , ) except KeyError : exc = traceback . format_exc ( ) if exc : return _fail_or_default ( ImportError ( exc ) , default ) mod = ProxyModule ( mod ) reloader = ModuleReloader ( proxy = mod , name = name , path = path , initial_globals = initial_globals , ) _reloaders [ mod ] = reloader threaded = False try : asyncio . get_running_loop ( ) except RuntimeError : threaded = True if not threaded : reloader . start_async ( ) else : reloader . start_threaded ( ) if not all ( isfunction ( value ) for key , value in mod . __dict__ . items ( ) if key not in initial_globals . keys ( ) and not key . startswith ( "__" ) ) : warn ( Message . not_reloadable ( name ) , NotReloadableWarning ) else : with open ( path , "rb" ) as rfile : code = rfile . read ( ) frame = inspect . getframeinfo ( inspect . currentframe ( ) ) self . _set_mod ( name = name , mod = mod , frame = frame ) try : mod = _build_mod ( name = name , code = code , initial_globals = initial_globals , module_path = path , ) except KeyError : del self . _using [ name ] exc = traceback . format_exc ( ) except KeyError : exc = traceback . format_exc ( ) finally : os . chdir ( original_cwd ) if exc : return _fail_or_default ( ImportError ( exc ) , default ) if as_import : sys . modules [ as_import ] = mod frame = inspect . getframeinfo ( inspect . currentframe ( ) ) self . _set_mod ( name = name , mod = mod , frame = frame ) return ProxyModule ( mod )
Import a module from a path.
Import a module from a path .
"""Import a module from a path.

        https://github.com/amogorkon/justuse/wiki/Use-Path

        Args:
            path ([type]): must be a pathlib.Path
            initial_globals ([type], optional): dict that should be globally available to the module before executing it. Defaults to None.
            default ([type], optional): Return instead if an exception is encountered.
            modes (int, optional): [description]. Defaults to 0; Acceptable mode for this variant: use.reloading.

        Returns:
            Optional[ModuleType]: The module if it was imported, otherwise whatever was specified as default.
        """ # calling from another use()d module # let's see where we started # there are a number of ways to call use() from a non-use() starting point # let's first check if we are running in jupyter # we're in jupyter, we use the CWD as set in the notebook # initial instance, if this doesn't work, just throw the towel # this looks like a hack, but isn't one - # jupyter is running an async loop internally, which works better async than threaded! # we're dealing with non-async code, we need threading # can't have the code inside the handler because of "during handling of X, another exception Y happened" # NOT reloading # the path needs to be set before attempting to load the new module - recursion confusing ftw! # let's not confuse the user and restore the cwd to the original in any case
The module if it was imported, otherwise whatever was specified as default.
The module if it was imported otherwise whatever was specified as default .
Optional[ModuleType]
self
path
must be a pathlib.Path
must be a pathlib . Path
initial_globals
dict that should be globally available to the module before executing it. Defaults to None.
dict that should be globally available to the module before executing it . Defaults to None .
None
as_import
str
default
Return instead if an exception is encountered.
Return instead if an exception is encountered .
modes
[description]. Defaults to 0; Acceptable mode for this variant: use.reloading.
[ description ] . Defaults to 0 ; Acceptable mode for this variant : use . reloading .
0; Acceptable mode for this variant: use.reloading