使用 Python 操作自定义 COM 接口
4.60/5 (11投票s)
开始使用 comtypes 包的分步教程。
目录
引言
互联网上有很多关于 Python 和 COM 的教程，但在实际操作中，一旦你开始接触标准 IDispatch 以外的内容，很容易就会感到困惑。当我想为我们的 COM 组件集编写单元测试时，我也遇到了同样的问题。这些组件相当简单，它们实现了一个自定义接口（派生自 IUnknown）和一个用于事件的传出 IDispatch 接口。
起初，我尝试使用标准的 pythoncom 模块，但发现它不支持自定义 COM 接口。然后，我下载了 comtypes 包并开始尝试使用它。由于文档匮乏，我花了一个晚上才写出了一个简单的示例。因此，这里是如何开始使用 comtypes 的分步指南。
编写 COM 对象
我们将编写一个 COM 组件，其中我们将使用 Python-COM 互操作性的基本技术和一些与线程相关的技巧。
我们在本教程中创建的组件公开了 ITaskLauncher 接口，并支持传出的 _ITaskLauncherEvents 接口。这是 IDL 文件中的一段摘录。
interface ITaskLauncher : IUnknown{ [id(1), helpstring("method StartTask")] HRESULT StartTask([in] BSTR name); }; dispinterface _ITaskLauncherEvents { methods: [id(1), helpstring("method TaskQueued")] HRESULT TaskQueued([in] BSTR name); [id(2), helpstring("method TaskCompleted")] HRESULT TaskCompleted([in] BSTR name); };
首先，创建一个 Visual Studio 项目。选择“ATL 项目”模板，为您的项目命名，然后单击“确定”。在“应用程序设置”页面上，将服务器类型设置为“可执行文件 (EXE)”，然后单击“完成”。
切换到类视图，选择新创建的项目，然后在上下文菜单中选择“添加”->“类...”。选择“ATL 简单对象”并单击“添加”。给它一个短名称“TaskLauncher”，然后保留所有其他字段，单击“下一步”。在“选项”页面上，将线程模型设置为“自由”，将接口设置为“自定义”，并选中“自动化兼容”复选框。另外，选中“连接点”以向您的类添加事件支持。单击“完成”以创建类。
重要提示：创建自定义接口对象时，应选中“自动化兼容”复选框。否则，脚本语言将无法访问您的接口。但是，您可以稍后通过直接修改 .idl 文件，将此属性命名为 oleautomation 来设置。
在类视图中，找到 ITaskLauncher 接口，然后在上下文菜单中选择“添加”->“添加方法...”。这将打开“添加方法向导”。将方法名称设置为“StartTask”，选中“in”属性，选择 BSTR 参数类型，将参数名称设置为“name”，然后单击“添加”。单击“完成”。在此步骤中，向导将创建一个 StartTask 方法，并在我们的 CTaskLauncher 类中实现此方法的一个空函数体。
在类视图中，找到 TaskServerLib，展开它，然后找到 _ITaskLauncherEvents 接口。在上下文菜单中，选择“添加”->“添加方法...”。将返回类型保留为 HRESULT，将方法名称设置为“TaskQueued”，并像之前一样添加“name”参数。单击“完成”。对“TaskCompleted”方法重复此操作，使用相同的参数。
现在，我们的声明事件的源接口已准备就绪，但我们需要 Visual Studio 来实现实际触发事件的函数。为此，在类视图中找到 CTaskLauncher 类，然后在上下文菜单中选择“添加”->“添加连接点...”。在出现的对话框中，双击 _ITaskLauncherEvents 接口，然后单击“完成”。
生成项目，以确保在此阶段没有错误。现在，我们已准备好实际实现组件方法。
打开 TaskLauncher.h 文件，并在末尾添加以下定义。
struct TaskInfo
{
BSTR name;
TaskInfo(BSTR taskName)
{
//copy taskName to name
UINT len = ::SysStringLen(taskName);
name = ::SysAllocStringLen(taskName, len);
}
~TaskInfo()
{
::SysFreeString(name);
}
};
在 TaskLauncher.cpp 中找到 StartTask 函数，并添加以下实现。
STDMETHODIMP CTaskLauncher::StartTask(BSTR name)
{
TaskInfo* pTaskInfo = new TaskInfo(name);
BSTR taskName = ::SysAllocStringLen(pTaskInfo->name,
::SysStringLen(pTaskInfo->name));
Fire_TaskQueued(taskName);
delete pTaskInfo;
return S_OK;
}
现在，这看起来有点复杂，但我们稍后将需要 TaskInfo 结构。现在，是时候构建我们的组件并开始编写客户端代码了。
用 Python 编写 COM 客户端
首先，我们需要知道我们类型库的 GUID。打开 Visual Studio 生成的 TaskServer.idl，找到下图所示的代码块。复制 uuid 属性的内容。
打开 PythonWin IDE，创建一个新的 Python 脚本，将 comtypes.GUID(...) 替换为您由 Visual Studio 生成的 GUID。
import comtypes.client as cc
import comtypes
tlb_id = comtypes.GUID("{3DED0EFB-21ED-4337-B098-1B8316952FFA}")
cc.GetModule((tlb_id, 1, 0))
import comtypes.gen.TaskServerLib as TaskLib
class Sink:
def TaskQueued(self, this, name):
print "TaskQueued event. name = %s" % name
def TaskCompleted(self, this, name):
print "TaskCompleted event. name = %s" % name
task_launcher = cc.CreateObject("TaskServer.TaskLauncher",
None, None, TaskLib.ITaskLauncher)
sink = Sink()
advise = cc.GetEvents(task_launcher, sink)
task_launcher.StartTask("first task")
cc.PumpEvents(5)
advise = None
task_launcher = None
在这里，我们通过调用 GetModule 来生成 TaskServerLib 模块，并将类型库 GUID、主库版本（1）和次版本（0）作为参数传递。接下来，我们声明 Sink 类，它将接收来自我们对象的事件。cc.CreateObject 创建一个 COM 对象并从中获取 ITaskLauncher 接口。此时，我们可以调用对象的方法，但要接收事件，我们需要一些额外的设置。创建一个 Sink 类实例，并调用 cc.GetEvents 将 task_launcher 源接口绑定到 sink。GetEvents 返回一个 advise 连接，最好保留对其的引用。否则，advise 连接可能会被垃圾回收，事件将停止工作。接下来，我们在 PumpEvents 循环中调用我们的方法 StartTask 并等待事件 5 秒钟。
运行此脚本。您的输出应如下所示：
# Generating comtypes.gen._3DED0EFB_21ED_4337_B098_1B8316952FFA_0_1_0
# Generating comtypes.gen._00020430_0000_0000_C000_000000000046_0_2_0
# Generating comtypes.gen.stdole
# Generating comtypes.gen.TaskServerLib
TaskQueued event. name = first task
线程间接口封送
现在，是时候修改我们的 COM 服务器使其更具异步性了。StartTask 方法在调用后立即触发 TaskQueued 事件。让我们添加一个工作线程，它将等待几秒钟并触发 TaskCompleted 事件。Visual Studio 为我们生成了 Fire_TaskCompleted 代理函数，但直接从工作线程调用它相当无用，因为它不会从工作线程到主线程进行接口封送。我猜在 ATL 中没有优雅的解决方案可以克服这个问题。我们可以修改 CProxy_ITaskLauncherEvents::Fire_TaskCompleted 函数并自己进行所有封送，但这样，如果我们的接口发生更改，我们就无法生成此文件。另一种方法是在我们的 ITaskLauncher 接口中引入 Fire_TaskCompletedInternal 方法，并将封送后的 ITaskLauncher 接口指针传递给工作线程。由于 Fire_TaskCompletedInternal 方法不应被 COM 客户端直接调用，因此我们将使其隐藏，尽管它仍将保留在 ITaskLauncher 虚函数表中。
因此，在类视图中，找到 ITaskLauncher 接口，在上下文菜单中选择“添加”->“添加方法...”。将方法名称填写为 Fire_TaskCompletedInternal，添加具有 [in] 方向和 BSTR 类型的“name”参数。单击“下一步”按钮或“IDL 属性”页面。选中“hidden”复选框，然后单击“完成”。
修改 TaskInfo 结构，并添加 LPSTREAM marshalledInterface 成员变量。
struct TaskInfo
{
BSTR name;
LPSTREAM marshalledInterface;
TaskInfo(BSTR taskName)
{
//copy taskName to name
UINT len = ::SysStringLen(taskName);
name = ::SysAllocStringLen(taskName, len);
}
~TaskInfo()
{
::SysFreeString(name);
}
};
找到 CTaskLauncher::StartTask 方法，并将其替换为以下代码。
STDMETHODIMP CTaskLauncher::StartTask(BSTR name)
{
TaskInfo* pTaskInfo = new TaskInfo(name);
BSTR taskName = ::SysAllocStringLen(pTaskInfo->name, ::SysStringLen(pTaskInfo->name));
Fire_TaskQueued(taskName);
CoMarshalInterThreadInterfaceInStream(IID_ITaskLauncher, (ITaskLauncher*)this,
&pTaskInfo->marshalledInterface);
if (_beginthreadex(NULL, 0, &threadFunc, (LPVOID)pTaskInfo, 0, NULL) == 0)
{
//clean up if we couldn't start the thread
pTaskInfo->marshalledInterface->Release();
delete pTaskInfo;
};
return S_OK;
}
将 threadFunc 函数定义插入到 StartTask 方法的正前方。
unsigned int __stdcall threadFunc(void* p)
{
CoInitializeEx(NULL, COINIT_MULTITHREADED);
Sleep(2000);
TaskInfo* pTaskInfo = (TaskInfo*)p;
ITaskLauncher* pTaskLauncher;
CoGetInterfaceAndReleaseStream(pTaskInfo->marshalledInterface,
IID_ITaskLauncher, (LPVOID*)&pTaskLauncher);
BSTR taskName = ::SysAllocStringLen(pTaskInfo->name, ::SysStringLen(pTaskInfo->name));
HRESULT hr = pTaskLauncher->Fire_TaskCompletedInternal(taskName);
delete pTaskInfo;
CoUninitialize();
return 0;
}
最后，找到 CTaskLauncher::Fire_TaskCompleteInternal 方法定义，并使其如下所示。
STDMETHODIMP CTaskLauncher::Fire_TaskCompletedInternal(BSTR name)
{
Fire_TaskCompleted(name);
return S_OK;
}
重新生成解决方案，然后再次尝试运行 Python 客户端。输出应如下所示：
# comtypes.gen._3DED0EFB_21ED_4337_B098_1B8316952FFA_0_1_0 must be regenerated
# Generating comtypes.gen._3DED0EFB_21ED_4337_B098_1B8316952FFA_0_1_0
# Generating comtypes.gen.TaskServerLib
TaskQueued event. name = first task
TaskCompleted event. name = first task
清单
- 为自定义接口设置
oleautomation属性。这对于 Python、VBA 等脚本语言通过 typelib 进行后期绑定是必需的。 - 在调用任何 COM 相关函数或接口函数之前，在每个线程中调用一次
CoInitializeEx。不要忘记在线程结束前调用CoUninitialize。 - 您应该在线程之间封送接口指针。有关更多详细信息，请参阅
CoMarshalInterThreadInterfaceInStream/CoGetInterfaceAndReleaseStream。另一种技术是“全局接口表”。有关更多详细信息，请参阅 何时使用全局接口表 文章。
下载本文代码
- TaskServer.zip (13K)
- TaskClient.zip (0.5K)
适用于 Visual C++ 2005 Express Edition 的解决方案、项目和源文件。
Python 脚本源文件。