使用字元串給定的名稱來導入模塊

我們已經有了需要導入的模塊名稱,但是這個名稱保存在一個字元串中。我們像在字元串上執行import命令。

當模塊和包的名稱以字元串的形式給出的時候,我們可以使用importlib.import_module()函數來手動導入這個模塊:

>>> import importlib
>>> math = importlib.import_module(math)
>>> math.sin(2)
0.9092974268256817
>>> mod = importlib.import_module(urllib.request)
>>> u = mod.urlopen(http://www.python.org)
>>>

importmodule基本上和import完成的步驟相同,但是import__module會把模塊對象作為結果返回給你。我們只需要將他保存在一個變數里,之後把它當作普通的模塊使用就好。

如果要同包打交道,import_module()也可以用來實現相對導入。但是,需要提供一個額外的參數:

import importlib
# Same as from . import b
b = importlib.import_module(.b, __package__)

使用import_module()手動導入模塊的問題通常出現在以某種方式編寫修改或覆蓋模塊的代碼時候。例如,也許你正在執行某種自定義導入機制,需要通過名稱來載入一個模塊,通過補丁載入代碼。

在舊的代碼,有時你會看到用於導入的內建函數__import__()。儘管它能工作,但是importlib.import_module() 通常更容易使用。

利用import鉤子從遠程機器上載入模塊

首先要提出來的是安全問題。本節討論的思想如果沒有一些額外的安全和認知機制的話會很糟糕。 也就是說,我們的主要目的是深入分析Python的import語句機制。 如果你理解了本節內部原理,你就能夠為其他任何目的而自定義import。 有了這些,讓我們繼續向前走。

本節核心是設計導入語句的擴展功能。有很多種方法可以做這個, 不過為了演示的方便,我們開始先構造下面這個Python代碼結構:

testcode/
spam.py
fib.py
grok/
__init__.py
blah.py

這些文件的內容並不重要,不過我們在每個文件中放入了少量的簡單語句和函數, 這樣你可以測試它們並查看當它們被導入時的輸出。例如:

# spam.py
print("Im spam")

def hello(name):
print(Hello %s % name)

# fib.py
print("Im fib")

def fib(n):
if n < 2:
return 1
else:
return fib(n-1) + fib(n-2)

# grok/__init__.py
print("Im grok.__init__")

# grok/blah.py
print("Im grok.blah")

這裡的目的是允許這些文件作為模塊被遠程訪問。 也許最簡單的方式就是將它們發布到一個web伺服器上面。在testcode目錄中像下面這樣運行Python:

bash % cd testcode
bash % python3 -m http.server 15000
Serving HTTP on 0.0.0.0 port 15000 ...

伺服器運行起來後再啟動一個單獨的Python解釋器。 確保你可以使用urllib訪問到遠程文件。例如:

>>> from urllib.request import urlopen
>>> u = urlopen(http://localhost:15000/fib.py)
>>> data = u.read().decode(utf-8)
>>> print(data)
# fib.py
print("Im fib")

def fib(n):
if n < 2:
return 1
else:
return fib(n-1) + fib(n-2)
>>>

從這個伺服器載入源代碼是接下來本節的基礎。 為了替代手動的通過 urlopen() 來收集源文件, 我們通過自定義import語句來在後台自動幫我們做到。

載入遠程模塊的第一種方法是創建一個顯式的載入函數來完成它。例如:

import imp
import urllib.request
import sys

def load_module(url):
u = urllib.request.urlopen(url)
source = u.read().decode(utf-8)
mod = sys.modules.setdefault(url, imp.new_module(url))
code = compile(source, url, exec)
mod.__file__ = url
mod.__package__ =
exec(code, mod.__dict__)
return mod

這個函數會下載源代碼,並使用compile()將其編譯到一個代碼對象中, 然後在一個新創建的模塊對象的字典中來執行它。一個更酷的做法是創建一個自定義導入器。第一種方法是創建一個元路徑導入器。如下:

# urlimport.py
import sys
import importlib.abc
import imp
from urllib.request import urlopen
from urllib.error import HTTPError, URLError
from html.parser import HTMLParser

# Debugging
import logging
log = logging.getLogger(__name__)

# Get links from a given URL
def _get_links(url):
class LinkParser(HTMLParser):
def handle_starttag(self, tag, attrs):
if tag == a:
attrs = dict(attrs)
links.add(attrs.get(href).rstrip(/))
links = set()
try:
log.debug(Getting links from %s % url)
u = urlopen(url)
parser = LinkParser()
parser.feed(u.read().decode(utf-8))
except Exception as e:
log.debug(Could not get links. %s, e)
log.debug(links: %r, links)
return links

class UrlMetaFinder(importlib.abc.MetaPathFinder):
def __init__(self, baseurl):
self._baseurl = baseurl
self._links = { }
self._loaders = { baseurl : UrlModuleLoader(baseurl) }

def find_module(self, fullname, path=None):
log.debug(find_module: fullname=%r, path=%r, fullname, path)
if path is None:
baseurl = self._baseurl
else:
if not path[0].startswith(self._baseurl):
return None
baseurl = path[0]
parts = fullname.split(.)
basename = parts[-1]
log.debug(find_module: baseurl=%r, basename=%r, baseurl, basename)

# Check link cache
if basename not in self._links:
self._links[baseurl] = _get_links(baseurl)

# Check if its a package
if basename in self._links[baseurl]:
log.debug(find_module: trying package %r, fullname)
fullurl = self._baseurl + / + basename
# Attempt to load the package (which accesses __init__.py)
loader = UrlPackageLoader(fullurl)
try:
loader.load_module(fullname)
self._links[fullurl] = _get_links(fullurl)
self._loaders[fullurl] = UrlModuleLoader(fullurl)
log.debug(find_module: package %r loaded, fullname)
except ImportError as e:
log.debug(find_module: package failed. %s, e)
loader = None
return loader
# A normal module
filename = basename + .py
if filename in self._links[baseurl]:
log.debug(find_module: module %r found, fullname)
return self._loaders[baseurl]
else:
log.debug(find_module: module %r not found, fullname)
return None

def invalidate_caches(self):
log.debug(invalidating link cache)
self._links.clear()

# Module Loader for a URL
class UrlModuleLoader(importlib.abc.SourceLoader):
def __init__(self, baseurl):
self._baseurl = baseurl
self._source_cache = {}

def module_repr(self, module):
return <urlmodule %r from %r> % (module.__name__, module.__file__)

# Required method
def load_module(self, fullname):
code = self.get_code(fullname)
mod = sys.modules.setdefault(fullname, imp.new_module(fullname))
mod.__file__ = self.get_filename(fullname)
mod.__loader__ = self
mod.__package__ = fullname.rpartition(.)[0]
exec(code, mod.__dict__)
return mod

# Optional extensions
def get_code(self, fullname):
src = self.get_source(fullname)
return compile(src, self.get_filename(fullname), exec)

def get_data(self, path):
pass

def get_filename(self, fullname):
return self._baseurl + / + fullname.split(.)[-1] + .py

def get_source(self, fullname):
filename = self.get_filename(fullname)
log.debug(loader: reading %r, filename)
if filename in self._source_cache:
log.debug(loader: cached %r, filename)
return self._source_cache[filename]
try:
u = urlopen(filename)
source = u.read().decode(utf-8)
log.debug(loader: %r loaded, filename)
self._source_cache[filename] = source
return source
except (HTTPError, URLError) as e:
log.debug(loader: %r failed. %s, filename, e)
raise ImportError("Cant load %s" % filename)

def is_package(self, fullname):
return False

# Package loader for a URL
class UrlPackageLoader(UrlModuleLoader):
def load_module(self, fullname):
mod = super().load_module(fullname)
mod.__path__ = [ self._baseurl ]
mod.__package__ = fullname

def get_filename(self, fullname):
return self._baseurl + / + __init__.py

def is_package(self, fullname):
return True

# Utility functions for installing/uninstalling the loader
_installed_meta_cache = { }
def install_meta(address):
if address not in _installed_meta_cache:
finder = UrlMetaFinder(address)
_installed_meta_cache[address] = finder
sys.meta_path.append(finder)
log.debug(%r installed on sys.meta_path, finder)

def remove_meta(address):
if address in _installed_meta_cache:
finder = _installed_meta_cache.pop(address)
sys.meta_path.remove(finder)
log.debug(%r removed from sys.meta_path, finder)

這個特殊的方案會安裝一個特別的查找器 UrlMetaFinder 實例, 作為 sys.meta_path 中最後的實體。 當模塊被導入時,會依據 sys.meta_path 中的查找器定位模塊。 在這個例子中,UrlMetaFinder 實例是最後一個查找器方案, 當模塊在任何一個普通地方都找不到的時候就觸發它。

作為常見的實現方案,UrlMetaFinder 類包裝在一個用戶指定的URL上。 在內部,查找器通過抓取指定URL的內容構建合法的鏈接集合。 導入的時候,模塊名會跟已有的鏈接作對比。如果找到了一個匹配的, 一個單獨的 UrlModuleLoader 類被用來從遠程機器上載入源代碼並創建最終的模塊對象。 這裡緩存鏈接的一個原因是避免不必要的HTTP請求重複導入。

自定義導入的第二種方法是編寫一個鉤子直接嵌入到 sys.path 變數中去, 識別某些目錄命名模式。 在 urlimport.py 中添加如下的類和支持函數:

# urlimport.py
# ... include previous code above ...
# Path finder class for a URL
class UrlPathFinder(importlib.abc.PathEntryFinder):
def __init__(self, baseurl):
self._links = None
self._loader = UrlModuleLoader(baseurl)
self._baseurl = baseurl

def find_loader(self, fullname):
log.debug(find_loader: %r, fullname)
parts = fullname.split(.)
basename = parts[-1]
# Check link cache
if self._links is None:
self._links = [] # See discussion
self._links = _get_links(self._baseurl)

# Check if its a package
if basename in self._links:
log.debug(find_loader: trying package %r, fullname)
fullurl = self._baseurl + / + basename
# Attempt to load the package (which accesses __init__.py)
loader = UrlPackageLoader(fullurl)
try:
loader.load_module(fullname)
log.debug(find_loader: package %r loaded, fullname)
except ImportError as e:
log.debug(find_loader: %r is a namespace package, fullname)
loader = None
return (loader, [fullurl])

# A normal module
filename = basename + .py
if filename in self._links:
log.debug(find_loader: module %r found, fullname)
return (self._loader, [])
else:
log.debug(find_loader: module %r not found, fullname)
return (None, [])

def invalidate_caches(self):
log.debug(invalidating link cache)
self._links = None

# Check path to see if it looks like a URL
_url_path_cache = {}
def handle_url(path):
if path.startswith((http://, https://)):
log.debug(Handle path? %s. [Yes], path)
if path in _url_path_cache:
finder = _url_path_cache[path]
else:
finder = UrlPathFinder(path)
_url_path_cache[path] = finder
return finder
else:
log.debug(Handle path? %s. [No], path)

def install_path_hook():
sys.path_hooks.append(handle_url)
sys.path_importer_cache.clear()
log.debug(Installing handle_url)

def remove_path_hook():
sys.path_hooks.remove(handle_url)
sys.path_importer_cache.clear()
log.debug(Removing handle_url)

事實上,一個用來在sys.path中查找URL的自定義路徑檢查函數已經構建完畢。 當它們被碰到的時候,一個新的 UrlPathFinder 實例被創建並被放入 sys.path_importer_cache. 之後,所有需要檢查 sys.path 的導入語句都會使用你的自定義查找器。

基於路徑導入的包處理稍微有點複雜,並且跟 find_loader() 方法返回值有關。 對於簡單模塊,find_loader() 返回一個元組(loader, None), 其中的loader是一個用於導入模塊的載入器實例。

對於一個普通的包,find_loader() 返回一個元組(loader, path), 其中的loader是一個用於導入包(並執行__init__.py)的載入器實例, path是一個會初始化包的 __path__ 屬性的目錄列表。 例如,如果基礎URL是 http://localhost:15000 並且一個用戶執行 import grok , 那麼 find_loader() 返回的path就會是 [ 『localhost:15000/grok』 ]

find_loader() 還要能處理一個命名空間包。 一個命名空間包中有一個合法的包目錄名,但是不存在__init__.py文件。 這樣的話,find_loader() 必須返回一個元組(None, path), path是一個目錄列表,由它來構建包的定義有__init__.py文件的__path__屬性。 對於這種情況,導入機制會繼續前行去檢查sys.path中的目錄。 如果找到了命名空間包,所有的結果路徑被加到一起來構建最終的命名空間包。

參考書目

《Python CookBook》作者:【美】 David Beazley, Brian K. Jones

Github地址:

yidao620c/python3-cookbook?

github.com
圖標

推薦閱讀:
相关文章