动态发现 python 模块和其中的 类
Dynamically discover python modules and classes within them
我正在 Python 3.7 中开发一个状态机，它将导入“Actions”（在它们自己的模块中定义为 classes）。我想要做的是让任何实现状态机的应用程序能够调用 import_actions() 方法并将包含它们的包的名称传递给它。
我还没有走到那一步，但我目前正在尝试将状态机所需的核心操作导入 bootstrap 本身，老实说我正在努力......
我有这个目录结构：
tree ism/core
ism/core
├── __init__.py
├── action.py
├── action_confirm_ready_to_run.py
├── action_emergency_shutdown.py
├── action_normal_shutdown.py
├── action_process_inbound_messages.py
├── data.json
└── schema.json
我的状态机在 ism/ 中，它试图导入的操作在 ism/core.
中
目前每个动作都是空的 class 像这样：
"""
from ism.core.action import Action
class ActionNormalShutdown(Action):
pass
我需要做的是：
动态发现这些文件 - 即使在这种情况下我可以看到它们，因为稍后在导入第三方操作时，我将不知道包中有什么。
导入它们，然后
发现其中 class 的名称（例如 ActionNormalShutdown），以便我可以实例化每个对象并将它们添加到集合中。
之后，状态机将无限循环集合并调用每个集合的 execute() 方法。
所以每个“动作包”都是一个 python 包，每个动作都是一个 class 在它自己的模块中。
在核心包的 init.py 中，我有这段代码可以动态创建 all 变量：
from os.path import dirname, basename, isfile, join
import glob
modules = glob.glob(join(dirname(__file__), "action*.py"))
__all__ = [ basename(f)[:-3] for f in modules if isfile(f) and not f.endswith('__init__.py')]
我的状态机中有这个方法，它似乎获得了 ism.core 包中的所有模块：
def __import_core_actions(self):
"""Import the core actions for the ISM"""
import importlib.util
core_actions = pkg_resources.contents('ism.core')
for action in core_actions:
for file in importlib.import_module('ism.core').modules:
spec = importlib.util.spec_from_file_location("MyAction", file)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
我认为这实际上是在“加载”模块。我把它放在引号中是因为我不确定加载是否与导入相同...
我不确定的另一件事 – 在 importlib.util.spec_from_file_location("MyAction", file)
的调用中
我使用 MyAction 的名称，主要是因为我需要一个名称参数，但我不知道为什么或是否应该使用它。所以任何帮助澄清，将不胜感激。
因此，如果该方法实际上是导入模块，那么我如何扩展代码以实例化在模块中找到的每个 class 并将每个实例添加到集合中？
展望未来，第三方开发人员可能会在本地系统上安装他们的操作包，并且可以将名称传递给我的导入方法，让我发现内容。所以我可以概括这个方法来接受一个包名，然后根据我自己的需要用 'ism.core' 调用它。
你能帮我进行动态反省吗？
包裹
有一个专门为此目的设计的包：setuptools.find_packages。
注意：您也可以使用 distutils.core.find_packages
如何使用
在包中使用时，非常简单：
首先，我们要导入它（和备用）:
try:
from setuptools import setup, find_packages
except ImportError:
from distutils.core import setup, find_packages
接下来，我们有一些通用的 setup.py 内容：
setup(
name='example_pkg',
version='1.2.3',
description='Example for Whosebug answer',
author='Person Human',
license='MIT',
（我知道这还不是全部）
然后，我们可以在包参数中放入：
packages=find_packages(),
)
这样就可以了。
我已经弄清楚了我想要达到的目标，为了清楚起见，任何人从搜索引擎中点击这个post，我决定post一个答案/对我自己的问题的解释。
对我试图实现的目标的更清晰的解释：
我正在开发一种状态机，它可以导入表达某些特定功能的操作“包”（即包）。
'Action' 将是一个 class，它继承自我的 BaseAction class，并且还表达了一个 'execute()' 方法。它还应该在其 class 名称中包含子字符串 'Action'。
这是我正在开发的一个示例包，它允许状态机对出现在入站目录中的基于入站文件的消息做出反应，以及通过出站目录发送基于文件的消息：
ism_comms/
├── __init__.py
├── file
│ ├── __init__.py
│ └── actions
│ ├── __init__.py
│ ├── action_io_file_before.py
│ ├── action_io_file_inbound_.py
│ ├── action_io_file_outbound.py
│ ├── data.json
│ └── schema.json
如您所见，ism_comms.file 包包含三个操作：
- action_io_file_before.py - 在状态机从 STARTING 切换到 运行 状态之前创建入站和出站目录。
- action_io_file_inbound - 监控消息的入站目录（json 文件）并将它们加载到控制数据库中的消息 table。
- action_io_file_outbound - 监视出站消息的消息 table 并将它们写入出站目录。
它们在自己的包中，还有 schema.json 和 data.json 个文件。
schema.json
{
"mysql": {
"tables": [
"CREATE TABLE IF NOT EXISTS messages ( message_id INTEGER NOT NULL AUTO_INCREMENT COMMENT 'Record ID in recipient messages table', sender TEXT NOT NULL COMMENT 'Return address of sender', sender_id INTEGER NOT NULL COMMENT 'Record ID in sender messages table', recipient TEXT NOT NULL COMMENT 'Address of recipient', action TEXT NOT NULL COMMENT 'Name of the action that handles this message', payload TEXT COMMENT 'Json body of msg payload', sent TEXT NOT NULL COMMENT 'Timestamp msg sent by sender', received TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT 'Time ism loaded message into database', direction TEXT NOT NULL COMMENT 'In or outbound message', processed BOOLEAN NOT NULL DEFAULT '0' COMMENT 'Has the message been processed?', PRIMARY KEY(id) );"
]
},
"sqlite3": {
"tables": [
"CREATE TABLE IF NOT EXISTS messages (\nmessage_id INTEGER NOT NULL PRIMARY KEY, -- Record ID in recipient messages table\nsender TEXT NOT NULL, -- Return address of sender\nsender_id INTEGER NOT NULL, -- Record ID in sender messages table\nrecipient TEXT NOT NULL, -- Address of recipient\naction TEXT NOT NULL, -- Name of the action that handles this message\npayload TEXT, -- Json body of msg payload\nsent TEXT NOT NULL, -- Timestamp msg sent by sender\nreceived TEXT NOT NULL DEFAULT (strftime('%s', 'now')), -- Timestamp ism loaded message into database\ndirection TEXT NOT NULL DEFAULT 'inbound', -- In or outbound message\nprocessed BOOLEAN NOT NULL DEFAULT '0' -- Has the message been processed\n);"
]
}
}
data.json
{
"mysql": {
"inserts": [
"INSERT INTO actions VALUES(NULL,'ActionIoFileBefore','STARTING','null',1)",
"INSERT INTO actions VALUES(NULL,'ActionIoFileInbound','RUNNING','null',0)",
"INSERT INTO actions VALUES(NULL,'ActionIoFileOutbound','RUNNING','null',0)"
]
},
"sqlite3": {
"inserts": [
"INSERT INTO actions VALUES(NULL,'ActionIoFileBefore','STARTING','null',1)",
"INSERT INTO actions VALUES(NULL,'ActionIoFileInbound','RUNNING','null',0)",
"INSERT INTO actions VALUES(NULL,'ActionIoFileOutbound','RUNNING','null',0)"
]
}
}
我想要达到的目标：
简而言之，我希望贡献者能够简单地通过将包名称传递给状态机来导入他们自己的动作包。
为此，我需要使用自省，但在 Python 中不熟悉如何使用。因此我 post 提出了上面的问题。现在我已经了解了如何使用 importlib 实现此目的，我将 post 代码放在这里。
任何关于如何使它变得更好的评论都非常受欢迎'Pythonic'，因为我对这门语言还比较陌生。
导入方式：
def import_action_pack(self, pack):
"""Import an action pack
Application can pass in action packs to enable the ISM to express
specific functionality. For an example of how to call this method,
see the unit test in tests/test_ism.py (test_import_action_pack).
Each action pack is a python package containing:
* At least one action class inheriting from ism.core.BaseAction
* A data.json file containing at least the insert statements for the
action in the control DB.
* Optionally a schema.json file contain the create statements for any
tables the action needs in the control DB.
The package should contain nothing else and no sub packages.
"""
import pkgutil
action_args = {
"dao": self.dao,
"properties": self.properties
}
try:
# Import the package containing the actions
package = importlib.import_module(pack)
# Find each action module in the package
for importer, modname, ispkg in pkgutil.iter_modules(package.__path__):
# Should not be any sub packages in there
if ispkg:
raise MalformedActionPack(
f'Passed malformed action pack ({pack}). Unexpected sub packages {modname}'
)
# Import the module containing the action
module = importlib.import_module(f'{pack}.{importer.find_spec(modname).name}')
# Get the name of the action class, instantiate it and add to the collection of actions
for action in inspect.getmembers(module, inspect.isclass):
if action[0] == 'BaseAction':
continue
if 'Action' in action[0]:
cl_ = getattr(module, action[0])
self.actions.append(cl_(action_args))
# Get the supporting DB file/s
self.import_action_pack_tables(package)
except ModuleNotFoundError as e:
logging.error(f'Module/s not found for argument ({pack})')
raise
架构的导入和SQL:
def import_action_pack_tables(self, package):
""""An action will typically create some tables and insert standing data.
If supporting schema file exists, then create the tables. A data.json
file must exist with at least one insert for the actions table or the action
execute method will not be able to activate or deactivate..
"""
inserts_found = False
path = os.path.split(package.__file__)[0]
for root, dirs, files in os.walk(path):
if 'schema.json' in files:
schema_file = os.path.join(root, 'schema.json')
with open(schema_file) as tables:
data = json.load(tables)
for table in data[self.properties['database']['rdbms'].lower()]['tables']:
self.dao.execute_sql_statement(table)
if 'data.json' in files:
data = os.path.join(root, 'data.json')
with open(data) as statements:
inserts = json.load(statements)
for insert in inserts[self.properties['database']['rdbms'].lower()]['inserts']:
self.dao.execute_sql_statement(insert)
inserts_found = True
if not inserts_found:
raise MalformedActionPack(f'No insert statements found for action pack ({package})')
我正在 Python 3.7 中开发一个状态机，它将导入“Actions”（在它们自己的模块中定义为 classes）。我想要做的是让任何实现状态机的应用程序能够调用 import_actions() 方法并将包含它们的包的名称传递给它。
我还没有走到那一步，但我目前正在尝试将状态机所需的核心操作导入 bootstrap 本身，老实说我正在努力......
我有这个目录结构：
tree ism/core
ism/core
├── __init__.py
├── action.py
├── action_confirm_ready_to_run.py
├── action_emergency_shutdown.py
├── action_normal_shutdown.py
├── action_process_inbound_messages.py
├── data.json
└── schema.json
我的状态机在 ism/ 中，它试图导入的操作在 ism/core.
中目前每个动作都是空的 class 像这样：
"""
from ism.core.action import Action
class ActionNormalShutdown(Action):
pass
我需要做的是：
动态发现这些文件 - 即使在这种情况下我可以看到它们，因为稍后在导入第三方操作时，我将不知道包中有什么。
导入它们，然后
发现其中 class 的名称（例如 ActionNormalShutdown），以便我可以实例化每个对象并将它们添加到集合中。
之后，状态机将无限循环集合并调用每个集合的 execute() 方法。
所以每个“动作包”都是一个 python 包，每个动作都是一个 class 在它自己的模块中。
在核心包的 init.py 中，我有这段代码可以动态创建 all 变量：
from os.path import dirname, basename, isfile, join
import glob
modules = glob.glob(join(dirname(__file__), "action*.py"))
__all__ = [ basename(f)[:-3] for f in modules if isfile(f) and not f.endswith('__init__.py')]
我的状态机中有这个方法，它似乎获得了 ism.core 包中的所有模块：
def __import_core_actions(self):
"""Import the core actions for the ISM"""
import importlib.util
core_actions = pkg_resources.contents('ism.core')
for action in core_actions:
for file in importlib.import_module('ism.core').modules:
spec = importlib.util.spec_from_file_location("MyAction", file)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
我认为这实际上是在“加载”模块。我把它放在引号中是因为我不确定加载是否与导入相同...
我不确定的另一件事 – 在 importlib.util.spec_from_file_location("MyAction", file)
我使用 MyAction 的名称，主要是因为我需要一个名称参数，但我不知道为什么或是否应该使用它。所以任何帮助澄清，将不胜感激。
因此，如果该方法实际上是导入模块，那么我如何扩展代码以实例化在模块中找到的每个 class 并将每个实例添加到集合中？
展望未来，第三方开发人员可能会在本地系统上安装他们的操作包，并且可以将名称传递给我的导入方法，让我发现内容。所以我可以概括这个方法来接受一个包名，然后根据我自己的需要用 'ism.core' 调用它。
你能帮我进行动态反省吗？
包裹
有一个专门为此目的设计的包：setuptools.find_packages。
注意：您也可以使用 distutils.core.find_packages
如何使用
在包中使用时，非常简单：
首先，我们要导入它（和备用）:
try:
from setuptools import setup, find_packages
except ImportError:
from distutils.core import setup, find_packages
接下来，我们有一些通用的 setup.py 内容：
setup(
name='example_pkg',
version='1.2.3',
description='Example for Whosebug answer',
author='Person Human',
license='MIT',
（我知道这还不是全部）
然后，我们可以在包参数中放入：
packages=find_packages(),
)
这样就可以了。
我已经弄清楚了我想要达到的目标，为了清楚起见，任何人从搜索引擎中点击这个post，我决定post一个答案/对我自己的问题的解释。
对我试图实现的目标的更清晰的解释：
我正在开发一种状态机，它可以导入表达某些特定功能的操作“包”（即包）。
'Action' 将是一个 class，它继承自我的 BaseAction class，并且还表达了一个 'execute()' 方法。它还应该在其 class 名称中包含子字符串 'Action'。
这是我正在开发的一个示例包，它允许状态机对出现在入站目录中的基于入站文件的消息做出反应，以及通过出站目录发送基于文件的消息：
ism_comms/
├── __init__.py
├── file
│ ├── __init__.py
│ └── actions
│ ├── __init__.py
│ ├── action_io_file_before.py
│ ├── action_io_file_inbound_.py
│ ├── action_io_file_outbound.py
│ ├── data.json
│ └── schema.json
如您所见，ism_comms.file 包包含三个操作：
- action_io_file_before.py - 在状态机从 STARTING 切换到 运行 状态之前创建入站和出站目录。
- action_io_file_inbound - 监控消息的入站目录（json 文件）并将它们加载到控制数据库中的消息 table。
- action_io_file_outbound - 监视出站消息的消息 table 并将它们写入出站目录。
它们在自己的包中，还有 schema.json 和 data.json 个文件。
schema.json
{
"mysql": {
"tables": [
"CREATE TABLE IF NOT EXISTS messages ( message_id INTEGER NOT NULL AUTO_INCREMENT COMMENT 'Record ID in recipient messages table', sender TEXT NOT NULL COMMENT 'Return address of sender', sender_id INTEGER NOT NULL COMMENT 'Record ID in sender messages table', recipient TEXT NOT NULL COMMENT 'Address of recipient', action TEXT NOT NULL COMMENT 'Name of the action that handles this message', payload TEXT COMMENT 'Json body of msg payload', sent TEXT NOT NULL COMMENT 'Timestamp msg sent by sender', received TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT 'Time ism loaded message into database', direction TEXT NOT NULL COMMENT 'In or outbound message', processed BOOLEAN NOT NULL DEFAULT '0' COMMENT 'Has the message been processed?', PRIMARY KEY(id) );"
]
},
"sqlite3": {
"tables": [
"CREATE TABLE IF NOT EXISTS messages (\nmessage_id INTEGER NOT NULL PRIMARY KEY, -- Record ID in recipient messages table\nsender TEXT NOT NULL, -- Return address of sender\nsender_id INTEGER NOT NULL, -- Record ID in sender messages table\nrecipient TEXT NOT NULL, -- Address of recipient\naction TEXT NOT NULL, -- Name of the action that handles this message\npayload TEXT, -- Json body of msg payload\nsent TEXT NOT NULL, -- Timestamp msg sent by sender\nreceived TEXT NOT NULL DEFAULT (strftime('%s', 'now')), -- Timestamp ism loaded message into database\ndirection TEXT NOT NULL DEFAULT 'inbound', -- In or outbound message\nprocessed BOOLEAN NOT NULL DEFAULT '0' -- Has the message been processed\n);"
]
}
}
data.json
{
"mysql": {
"inserts": [
"INSERT INTO actions VALUES(NULL,'ActionIoFileBefore','STARTING','null',1)",
"INSERT INTO actions VALUES(NULL,'ActionIoFileInbound','RUNNING','null',0)",
"INSERT INTO actions VALUES(NULL,'ActionIoFileOutbound','RUNNING','null',0)"
]
},
"sqlite3": {
"inserts": [
"INSERT INTO actions VALUES(NULL,'ActionIoFileBefore','STARTING','null',1)",
"INSERT INTO actions VALUES(NULL,'ActionIoFileInbound','RUNNING','null',0)",
"INSERT INTO actions VALUES(NULL,'ActionIoFileOutbound','RUNNING','null',0)"
]
}
}
我想要达到的目标：
简而言之，我希望贡献者能够简单地通过将包名称传递给状态机来导入他们自己的动作包。
为此，我需要使用自省，但在 Python 中不熟悉如何使用。因此我 post 提出了上面的问题。现在我已经了解了如何使用 importlib 实现此目的，我将 post 代码放在这里。
任何关于如何使它变得更好的评论都非常受欢迎'Pythonic'，因为我对这门语言还比较陌生。
导入方式：
def import_action_pack(self, pack):
"""Import an action pack
Application can pass in action packs to enable the ISM to express
specific functionality. For an example of how to call this method,
see the unit test in tests/test_ism.py (test_import_action_pack).
Each action pack is a python package containing:
* At least one action class inheriting from ism.core.BaseAction
* A data.json file containing at least the insert statements for the
action in the control DB.
* Optionally a schema.json file contain the create statements for any
tables the action needs in the control DB.
The package should contain nothing else and no sub packages.
"""
import pkgutil
action_args = {
"dao": self.dao,
"properties": self.properties
}
try:
# Import the package containing the actions
package = importlib.import_module(pack)
# Find each action module in the package
for importer, modname, ispkg in pkgutil.iter_modules(package.__path__):
# Should not be any sub packages in there
if ispkg:
raise MalformedActionPack(
f'Passed malformed action pack ({pack}). Unexpected sub packages {modname}'
)
# Import the module containing the action
module = importlib.import_module(f'{pack}.{importer.find_spec(modname).name}')
# Get the name of the action class, instantiate it and add to the collection of actions
for action in inspect.getmembers(module, inspect.isclass):
if action[0] == 'BaseAction':
continue
if 'Action' in action[0]:
cl_ = getattr(module, action[0])
self.actions.append(cl_(action_args))
# Get the supporting DB file/s
self.import_action_pack_tables(package)
except ModuleNotFoundError as e:
logging.error(f'Module/s not found for argument ({pack})')
raise
架构的导入和SQL:
def import_action_pack_tables(self, package):
""""An action will typically create some tables and insert standing data.
If supporting schema file exists, then create the tables. A data.json
file must exist with at least one insert for the actions table or the action
execute method will not be able to activate or deactivate..
"""
inserts_found = False
path = os.path.split(package.__file__)[0]
for root, dirs, files in os.walk(path):
if 'schema.json' in files:
schema_file = os.path.join(root, 'schema.json')
with open(schema_file) as tables:
data = json.load(tables)
for table in data[self.properties['database']['rdbms'].lower()]['tables']:
self.dao.execute_sql_statement(table)
if 'data.json' in files:
data = os.path.join(root, 'data.json')
with open(data) as statements:
inserts = json.load(statements)
for insert in inserts[self.properties['database']['rdbms'].lower()]['inserts']:
self.dao.execute_sql_statement(insert)
inserts_found = True
if not inserts_found:
raise MalformedActionPack(f'No insert statements found for action pack ({package})')