Implement PEP 338 which has been marked as accepted by GvR

This commit is contained in:
Nick Coghlan 2006-03-15 11:00:26 +00:00
parent 8ea61f1a83
commit e2ebb2d7f7
3 changed files with 625 additions and 42 deletions

431
Lib/runpy.py Executable file
View File

@ -0,0 +1,431 @@
"""runpy.py - locating and running Python code using the module namespace
Provides support for locating and running Python scripts using the Python
module namespace instead of the native filesystem.
This allows Python code to play nicely with non-filesystem based PEP 302
importers when locating support scripts as well as when importing modules.
"""
# Written by Nick Coghlan <ncoghlan at gmail.com>
# to implement PEP 338 (Executing Modules as Scripts)
import sys
import imp
__all__ = [
"run_module",
]
try:
_get_loader = imp.get_loader
except AttributeError:
# get_loader() is not provided by the imp module, so emulate it
# as best we can using the PEP 302 import machinery exposed since
# Python 2.3. The emulation isn't perfect, but the differences
# in the way names are shadowed shouldn't matter in practice.
import os.path
import marshal # Handle compiled Python files
# This helper is needed in order for the PEP 302 emulation to
# correctly handle compiled files
def _read_compiled_file(compiled_file):
magic = compiled_file.read(4)
if magic != imp.get_magic():
return None
try:
compiled_file.read(4) # Skip timestamp
return marshal.load(compiled_file)
except Exception:
return None
class _AbsoluteImporter(object):
"""PEP 302 importer wrapper for top level import machinery"""
def find_module(self, mod_name, path=None):
if path is not None:
return None
try:
file, filename, mod_info = imp.find_module(mod_name)
except ImportError:
return None
suffix, mode, mod_type = mod_info
if mod_type == imp.PY_SOURCE:
loader = _SourceFileLoader(mod_name, file,
filename, mod_info)
elif mod_type == imp.PY_COMPILED:
loader = _CompiledFileLoader(mod_name, file,
filename, mod_info)
elif mod_type == imp.PKG_DIRECTORY:
loader = _PackageDirLoader(mod_name, file,
filename, mod_info)
elif mod_type == imp.C_EXTENSION:
loader = _FileSystemLoader(mod_name, file,
filename, mod_info)
else:
loader = _BasicLoader(mod_name, file,
filename, mod_info)
return loader
class _FileSystemImporter(object):
"""PEP 302 importer wrapper for filesystem based imports"""
def __init__(self, path_item=None):
if path_item is not None:
if path_item != '' and not os.path.isdir(path_item):
raise ImportError("%s is not a directory" % path_item)
self.path_dir = path_item
else:
raise ImportError("Filesystem importer requires "
"a directory name")
def find_module(self, mod_name, path=None):
if path is not None:
return None
path_dir = self.path_dir
if path_dir == '':
path_dir = os.getcwd()
sub_name = mod_name.rsplit(".", 1)[-1]
try:
file, filename, mod_info = imp.find_module(sub_name,
[path_dir])
except ImportError:
return None
if not filename.startswith(path_dir):
return None
suffix, mode, mod_type = mod_info
if mod_type == imp.PY_SOURCE:
loader = _SourceFileLoader(mod_name, file,
filename, mod_info)
elif mod_type == imp.PY_COMPILED:
loader = _CompiledFileLoader(mod_name, file,
filename, mod_info)
elif mod_type == imp.PKG_DIRECTORY:
loader = _PackageDirLoader(mod_name, file,
filename, mod_info)
elif mod_type == imp.C_EXTENSION:
loader = _FileSystemLoader(mod_name, file,
filename, mod_info)
else:
loader = _BasicLoader(mod_name, file,
filename, mod_info)
return loader
class _BasicLoader(object):
"""PEP 302 loader wrapper for top level import machinery"""
def __init__(self, mod_name, file, filename, mod_info):
self.mod_name = mod_name
self.file = file
self.filename = filename
self.mod_info = mod_info
def _fix_name(self, mod_name):
if mod_name is None:
mod_name = self.mod_name
elif mod_name != self.mod_name:
raise ImportError("Loader for module %s cannot handle "
"module %s" % (self.mod_name, mod_name))
return mod_name
def load_module(self, mod_name=None):
mod_name = self._fix_name(mod_name)
mod = imp.load_module(mod_name, self.file,
self.filename, self.mod_info)
mod.__loader__ = self # for introspection
return mod
def get_code(self, mod_name=None):
return None
def get_source(self, mod_name=None):
return None
def is_package(self, mod_name=None):
return False
def close(self):
if self.file:
self.file.close()
def __del__(self):
self.close()
class _FileSystemLoader(_BasicLoader):
"""PEP 302 loader wrapper for filesystem based imports"""
def get_code(self, mod_name=None):
mod_name = self._fix_name(mod_name)
return self._get_code(mod_name)
def get_data(self, pathname):
return open(pathname, "rb").read()
def get_filename(self, mod_name=None):
mod_name = self._fix_name(mod_name)
return self._get_filename(mod_name)
def get_source(self, mod_name=None):
mod_name = self._fix_name(mod_name)
return self._get_source(mod_name)
def is_package(self, mod_name=None):
mod_name = self._fix_name(mod_name)
return self._is_package(mod_name)
def _get_code(self, mod_name):
return None
def _get_filename(self, mod_name):
return self.filename
def _get_source(self, mod_name):
return None
def _is_package(self, mod_name):
return False
class _PackageDirLoader(_FileSystemLoader):
"""PEP 302 loader wrapper for PKG_DIRECTORY directories"""
def _is_package(self, mod_name):
return True
class _SourceFileLoader(_FileSystemLoader):
"""PEP 302 loader wrapper for PY_SOURCE modules"""
def _get_code(self, mod_name):
return compile(self._get_source(mod_name),
self.filename, 'exec')
def _get_source(self, mod_name):
f = self.file
f.seek(0)
return f.read()
class _CompiledFileLoader(_FileSystemLoader):
"""PEP 302 loader wrapper for PY_COMPILED modules"""
def _get_code(self, mod_name):
f = self.file
f.seek(0)
return _read_compiled_file(f)
def _get_importer(path_item):
"""Retrieve a PEP 302 importer for the given path item
The returned importer is cached in sys.path_importer_cache
if it was newly created by a path hook.
If there is no importer, a wrapper around the basic import
machinery is returned. This wrapper is never inserted into
the importer cache (None is inserted instead).
The cache (or part of it) can be cleared manually if a
rescan of sys.path_hooks is necessary.
"""
try:
importer = sys.path_importer_cache[path_item]
except KeyError:
for path_hook in sys.path_hooks:
try:
importer = path_hook(path_item)
break
except ImportError:
pass
else:
importer = None
sys.path_importer_cache[path_item] = importer
if importer is None:
try:
importer = _FileSystemImporter(path_item)
except ImportError:
pass
return importer
def _get_path_loader(mod_name, path=None):
"""Retrieve a PEP 302 loader using a path importer"""
if path is None:
path = sys.path
absolute_loader = _AbsoluteImporter().find_module(mod_name)
if isinstance(absolute_loader, _FileSystemLoader):
# Found in filesystem, so scan path hooks
# before accepting this one as the right one
loader = None
else:
# Not found in filesystem, so use top-level loader
loader = absolute_loader
else:
loader = absolute_loader = None
if loader is None:
for path_item in path:
importer = _get_importer(path_item)
if importer is not None:
loader = importer.find_module(mod_name)
if loader is not None:
# Found a loader for our module
break
else:
# No path hook found, so accept the top level loader
loader = absolute_loader
return loader
def _get_package(pkg_name):
"""Retrieve a named package"""
pkg = __import__(pkg_name)
sub_pkg_names = pkg_name.split(".")
for sub_pkg in sub_pkg_names[1:]:
pkg = getattr(pkg, sub_pkg)
return pkg
def _get_loader(mod_name, path=None):
"""Retrieve a PEP 302 loader for the given module or package
If the module or package is accessible via the normal import
mechanism, a wrapper around the relevant part of that machinery
is returned.
Non PEP 302 mechanisms (e.g. the Windows registry) used by the
standard import machinery to find files in alternative locations
are partially supported, but are searched AFTER sys.path. Normally,
these locations are searched BEFORE sys.path, preventing sys.path
entries from shadowing them.
For this to cause a visible difference in behaviour, there must
be a module or package name that is accessible via both sys.path
and one of the non PEP 302 file system mechanisms. In this case,
the emulation will find the former version, while the builtin
import mechanism will find the latter.
Items of the following types can be affected by this discrepancy:
imp.C_EXTENSION
imp.PY_SOURCE
imp.PY_COMPILED
imp.PKG_DIRECTORY
"""
try:
loader = sys.modules[mod_name].__loader__
except (KeyError, AttributeError):
loader = None
if loader is None:
imp.acquire_lock()
try:
# Module not in sys.modules, or uses an unhooked loader
parts = mod_name.rsplit(".", 1)
if len(parts) == 2:
# Sub package, so use parent package's path
pkg_name, sub_name = parts
if pkg_name and pkg_name[0] != '.':
if path is not None:
raise ImportError("Path argument must be None "
"for a dotted module name")
pkg = _get_package(pkg_name)
try:
path = pkg.__path__
except AttributeError:
raise ImportError(pkg_name +
" is not a package")
else:
raise ImportError("Relative import syntax is not "
"supported by _get_loader()")
else:
# Top level module, so stick with default path
sub_name = mod_name
for importer in sys.meta_path:
loader = importer.find_module(mod_name, path)
if loader is not None:
# Found a metahook to handle the module
break
else:
# Handling via the standard path mechanism
loader = _get_path_loader(mod_name, path)
finally:
imp.release_lock()
return loader
# This helper is needed due to a missing component in the PEP 302
# loader protocol (specifically, "get_filename" is non-standard)
def _get_filename(loader, mod_name):
try:
get_filename = loader.get_filename
except AttributeError:
return None
else:
return get_filename(mod_name)
# ------------------------------------------------------------
# Done with the import machinery emulation, on with the code!
def _run_code(code, run_globals, init_globals,
mod_name, mod_fname, mod_loader):
"""Helper for _run_module_code"""
if init_globals is not None:
run_globals.update(init_globals)
run_globals.update(__name__ = mod_name,
__file__ = mod_fname,
__loader__ = mod_loader)
exec code in run_globals
return run_globals
def _run_module_code(code, init_globals=None,
mod_name=None, mod_fname=None,
mod_loader=None, alter_sys=False):
"""Helper for run_module"""
# Set up the top level namespace dictionary
if alter_sys:
# Modify sys.argv[0] and sys.module[mod_name]
temp_module = imp.new_module(mod_name)
mod_globals = temp_module.__dict__
saved_argv0 = sys.argv[0]
restore_module = mod_name in sys.modules
if restore_module:
saved_module = sys.modules[mod_name]
imp.acquire_lock()
try:
sys.argv[0] = mod_fname
sys.modules[mod_name] = temp_module
try:
_run_code(code, mod_globals, init_globals,
mod_name, mod_fname, mod_loader)
finally:
sys.argv[0] = saved_argv0
if restore_module:
sys.modules[mod_name] = saved_module
else:
del sys.modules[mod_name]
finally:
imp.release_lock()
# Copy the globals of the temporary module, as they
# may be cleared when the temporary module goes away
return mod_globals.copy()
else:
# Leave the sys module alone
return _run_code(code, {}, init_globals,
mod_name, mod_fname, mod_loader)
def run_module(mod_name, init_globals=None,
run_name=None, alter_sys=False):
"""Execute a module's code without importing it
Returns the resulting top level namespace dictionary
"""
loader = _get_loader(mod_name)
if loader is None:
raise ImportError("No module named " + mod_name)
code = loader.get_code(mod_name)
if code is None:
raise ImportError("No code object available for " + mod_name)
filename = _get_filename(loader, mod_name)
if run_name is None:
run_name = mod_name
return _run_module_code(code, init_globals, run_name,
filename, loader, alter_sys)
if __name__ == "__main__":
# Run the module specified as the next command line argument
if len(sys.argv) < 2:
print >> sys.stderr, "No module specified for execution"
else:
del sys.argv[0] # Make the requested module sys.argv[0]
run_module(sys.argv[0], run_name="__main__", alter_sys=True)

157
Lib/test/test_runpy.py Normal file
View File

@ -0,0 +1,157 @@
# Test the runpy module
import unittest
import os
import os.path
import sys
import tempfile
from test.test_support import verbose, run_unittest
from runpy import _run_module_code, run_module
# Set up the test code and expected results
class RunModuleCodeTest(unittest.TestCase):
expected_result = ["Top level assignment", "Lower level reference"]
test_source = (
"# Check basic code execution\n"
"result = ['Top level assignment']\n"
"def f():\n"
" result.append('Lower level reference')\n"
"f()\n"
"# Check the sys module\n"
"import sys\n"
"run_argv0 = sys.argv[0]\n"
"if __name__ in sys.modules:\n"
" run_name = sys.modules[__name__].__name__\n"
"# Check nested operation\n"
"import runpy\n"
"nested = runpy._run_module_code('x=1\\n', mod_name='<run>',\n"
" alter_sys=True)\n"
)
def test_run_module_code(self):
initial = object()
name = "<Nonsense>"
file = "Some other nonsense"
loader = "Now you're just being silly"
d1 = dict(initial=initial)
saved_argv0 = sys.argv[0]
d2 = _run_module_code(self.test_source,
d1,
name,
file,
loader,
True)
self.failUnless("result" not in d1)
self.failUnless(d2["initial"] is initial)
self.failUnless(d2["result"] == self.expected_result)
self.failUnless(d2["nested"]["x"] == 1)
self.failUnless(d2["__name__"] is name)
self.failUnless(d2["run_name"] is name)
self.failUnless(d2["__file__"] is file)
self.failUnless(d2["run_argv0"] is file)
self.failUnless(d2["__loader__"] is loader)
self.failUnless(sys.argv[0] is saved_argv0)
self.failUnless(name not in sys.modules)
def test_run_module_code_defaults(self):
saved_argv0 = sys.argv[0]
d = _run_module_code(self.test_source)
self.failUnless(d["result"] == self.expected_result)
self.failUnless(d["__name__"] is None)
self.failUnless(d["__file__"] is None)
self.failUnless(d["__loader__"] is None)
self.failUnless(d["run_argv0"] is saved_argv0)
self.failUnless("run_name" not in d)
self.failUnless(sys.argv[0] is saved_argv0)
class RunModuleTest(unittest.TestCase):
def expect_import_error(self, mod_name):
try:
run_module(mod_name)
except ImportError:
pass
else:
self.fail("Expected import error for " + mod_name)
def test_invalid_names(self):
self.expect_import_error("sys")
self.expect_import_error("sys.imp.eric")
self.expect_import_error("os.path.half")
self.expect_import_error("a.bee")
self.expect_import_error(".howard")
self.expect_import_error("..eaten")
def test_library_module(self):
run_module("runpy")
def _make_pkg(self, source, depth):
pkg_name = "__runpy_pkg__"
init_fname = "__init__"+os.extsep+"py"
test_fname = "runpy_test"+os.extsep+"py"
pkg_dir = sub_dir = tempfile.mkdtemp()
if verbose: print " Package tree in:", sub_dir
sys.path.insert(0, pkg_dir)
if verbose: print " Updated sys.path:", sys.path[0]
for i in range(depth):
sub_dir = os.path.join(sub_dir, pkg_name)
os.mkdir(sub_dir)
if verbose: print " Next level in:", sub_dir
pkg_fname = os.path.join(sub_dir, init_fname)
pkg_file = open(pkg_fname, "w")
pkg_file.write("__path__ = ['%s']\n" % sub_dir)
pkg_file.close()
if verbose: print " Created:", pkg_fname
mod_fname = os.path.join(sub_dir, test_fname)
mod_file = open(mod_fname, "w")
mod_file.write(source)
mod_file.close()
if verbose: print " Created:", mod_fname
mod_name = (pkg_name+".")*depth + "runpy_test"
return pkg_dir, mod_fname, mod_name
def _del_pkg(self, top, depth, mod_name):
for root, dirs, files in os.walk(top, topdown=False):
for name in files:
os.remove(os.path.join(root, name))
for name in dirs:
os.rmdir(os.path.join(root, name))
os.rmdir(top)
if verbose: print " Removed package tree"
for i in range(depth+1): # Don't forget the module itself
parts = mod_name.rsplit(".", i)
entry = parts[0]
del sys.modules[entry]
if verbose: print " Removed sys.modules entries"
del sys.path[0]
if verbose: print " Removed sys.path entry"
def _check_module(self, depth):
pkg_dir, mod_fname, mod_name = (
self._make_pkg("x=1\n", depth))
try:
if verbose: print "Running from source:", mod_name
d1 = run_module(mod_name) # Read from source
__import__(mod_name)
os.remove(mod_fname)
if verbose: print "Running from compiled:", mod_name
d2 = run_module(mod_name) # Read from bytecode
finally:
self._del_pkg(pkg_dir, depth, mod_name)
self.failUnless(d1["x"] == d2["x"] == 1)
if verbose: print "Module executed successfully"
def test_run_module(self):
for depth in range(4):
if verbose: print "Testing package depth:", depth
self._check_module(depth)
def test_main():
run_unittest(RunModuleCodeTest)
run_unittest(RunModuleTest)
if __name__ == "__main__":
test_main()

View File

@ -132,27 +132,42 @@ static void RunStartupFile(PyCompilerFlags *cf)
}
}
/* Get the path to a top-level module */
static struct filedescr * FindModule(const char *module,
FILE **fp, char **filename)
static int RunModule(char *module)
{
struct filedescr *fdescr = NULL;
*fp = NULL;
*filename = malloc(MAXPATHLEN);
if (*filename == NULL)
return NULL;
/* Find the actual module source code */
fdescr = _PyImport_FindModule(module, NULL,
*filename, MAXPATHLEN, fp, NULL);
if (fdescr == NULL) {
free(*filename);
*filename = NULL;
PyObject *runpy, *runmodule, *runargs, *result;
runpy = PyImport_ImportModule("runpy");
if (runpy == NULL) {
fprintf(stderr, "Could not import runpy module\n");
return -1;
}
return fdescr;
runmodule = PyObject_GetAttrString(runpy, "run_module");
if (runmodule == NULL) {
fprintf(stderr, "Could not access runpy.run_module\n");
Py_DECREF(runpy);
return -1;
}
runargs = Py_BuildValue("sOsO", module,
Py_None, "__main__", Py_True);
if (runargs == NULL) {
fprintf(stderr,
"Could not create arguments for runpy.run_module\n");
Py_DECREF(runpy);
Py_DECREF(runmodule);
return -1;
}
result = PyObject_Call(runmodule, runargs, NULL);
if (result == NULL) {
PyErr_Print();
}
Py_DECREF(runpy);
Py_DECREF(runmodule);
Py_DECREF(runargs);
if (result == NULL) {
return -1;
}
Py_DECREF(result);
return 0;
}
/* Main program */
@ -441,28 +456,9 @@ Py_Main(int argc, char **argv)
}
if (module != NULL) {
/* Backup _PyOS_optind and find the real file */
struct filedescr *fdescr = NULL;
/* Backup _PyOS_optind and force sys.arv[0] = module */
_PyOS_optind--;
if ((fdescr = FindModule(module, &fp, &filename))) {
argv[_PyOS_optind] = filename;
} else {
fprintf(stderr, "%s: module %s not found\n",
argv[0], module);
return 2;
}
if (!fp) {
fprintf(stderr,
"%s: module %s has no associated file\n",
argv[0], module);
return 2;
}
if (!_PyImport_IsScript(fdescr)) {
fprintf(stderr,
"%s: module %s not usable as script\n (%s)\n",
argv[0], module, filename);
return 2;
}
argv[_PyOS_optind] = module;
}
PySys_SetArgv(argc-_PyOS_optind, argv+_PyOS_optind);
@ -481,9 +477,8 @@ Py_Main(int argc, char **argv)
sts = PyRun_SimpleStringFlags(command, &cf) != 0;
free(command);
} else if (module) {
sts = PyRun_AnyFileExFlags(fp, filename, 1, &cf) != 0;
sts = RunModule(module);
free(module);
free(filename);
}
else {
if (filename == NULL && stdin_is_interactive) {