COM servers from within a Python service (on Win2k)
Duncan Booth
duncan at NOSPAMrcp.co.uk
Wed Dec 5 10:49:14 EST 2001
Syver Enstad <syver-en+usenet at online.no> wrote in news:u4rn5vh0x.fsf at online.no:
> Kragen Sitaker <kragen at pobox.com> writes:
>
>> Duncan Booth <duncan at NOSPAMrcp.co.uk> writes:
>> > Andrew Bennetts <andrew-pythonlist at puzzling.org> wrote in
>> > news:mailman.1007522528.18068.python-list at python.org:
>> > > Can anyone show me how to make this work?
>> >
>> > I have emailed you a working example. It is a minimal service which
>> > contains a COM server.
>>
>> Can you post it?
>>
>
> Yess, yess, preciousss, post for me too. Gollum.
>
Strange, last time this was discussed nobody was very interested.
Split up on the begin/end lines. The only thing to watch is that
you must use DCOMCNFG.EXE to set the default security to include
access by your userid otherwise access to the COM object will be
denied. Then run servicetest.py to check it all works.
The most likely problem will be some of the long lines wrapping.
Dont forget to change the guid if you use this for anything.
BTW, I suspect the QS_ALLEVENTS value should actually be QS_ALLINPUT.
---- begin readme.txt ----
This is a simple NT/2000 service that exposes a COM object.
servicetest.py runs unit tests including installing and removing the
service. It leaves the service installed and running on completion.
Alternatively install the service with:
service.py install
You must start the service to register the COM object:
service.py start
Stopping the service (service.py stop) does not uninstall the COM
object. You must remove the service (service.py remove) to unregister
the COM object.
When the service is running it writes stdout and stderr to
logfile.txt, but these may be buffered so output won't necessarily
appear immediately.
Creating the exposed object will start the service. Destroying the
object doesn't stop the service (although it could be made to do this
quite easily).
To change the object that is exposed, edit COMNAME in service.py.
---- end readme.txt ----
---- begin servicetest.py ----
import unittest,weakref,types,os,sys,time,re
import service
import win32service, win32serviceutil,pywintypes
import win32com.client
from win32service import SERVICE_STOPPED, SERVICE_START_PENDING, SERVICE_STOP_PENDING, \
SERVICE_RUNNING, SERVICE_CONTINUE_PENDING, SERVICE_PAUSE_PENDING, SERVICE_PAUSED
SERVICENAME = service.MyService._svc_name_
COMNAME = service.COMOBJECT._reg_progid_
SERVICE_STATES = {
SERVICE_STOPPED: "Stopped",
SERVICE_START_PENDING: "Starting",
SERVICE_STOP_PENDING: "Stopping",
SERVICE_RUNNING: "Running",
SERVICE_CONTINUE_PENDING: "Continuing",
SERVICE_PAUSE_PENDING: "Pausing",
SERVICE_PAUSED: "Paused",
}
class ServiceStatus:
def __init__(self):
self.serviceType, self.currentState, self.controlsAccepted, \
self.win32ExitCode, self.serviceSpecificExitCode, self.checkPoint, \
self.waitHint = win32serviceutil.QueryServiceStatus(SERVICENAME)
# Uninstalling a service is asynchronous, so after uninstalling we might
# get a service status. If so sleep a bit and try again.
def isServiceInstalled(expected=None):
def installed():
try:
s = ServiceStatus()
except pywintypes.error, (code, fn, msg):
if code==1060: return 0 # Not installed.
raise
return 1
for i in range(10):
inst = installed()
if expected is None or inst==expected:
break
time.sleep(0.05)
return inst
def serviceState():
"""Returns service state SERVICE_STOPPED, SERVICE_RUNNING, or SERVICE_PAUSED.
If the service is changing state it waits up to 10 seconds for the state change to complete."""
maxtime = time.time() + 10
while time.time() < maxtime:
s = ServiceStatus() # Raises exception if not installed.
if s.currentState in (SERVICE_START_PENDING, SERVICE_STOP_PENDING, \
SERVICE_CONTINUE_PENDING, SERVICE_PAUSE_PENDING):
time.sleep(0.05)
continue
return s.currentState
def commandWithOutput(command, expected):
import tempfile
tmpfile = tempfile.mktemp()
try:
cmdLine = '"%s" %s>%s' % (sys.executable, command , tmpfile)
os.system(cmdLine)
output = open(tmpfile, "r").read()
if not output.find(expected) >= 0:
print "Expected:",`expected`
print "Got:",`output`
assert output.find(expected) >= 0
finally:
os.unlink(tmpfile)
def serviceStart():
if not serviceState()==SERVICE_RUNNING:
commandWithOutput("service.py start", "Starting service %s\n" % SERVICENAME)
assert serviceState()==SERVICE_RUNNING
def serviceInstall():
if not isServiceInstalled():
commandWithOutput("service.py install", "\nService installed\n")
assert isServiceInstalled(1), "Service should now be installed"
def serviceStop():
if serviceState() != SERVICE_STOPPED:
commandWithOutput("service.py stop", "Stopping service %s\n" % SERVICENAME)
assert serviceState()==SERVICE_STOPPED
def serviceRemove():
if isServiceInstalled(1):
commandWithOutput("service.py remove", "Removing service %s\n" % SERVICENAME)
assert not isServiceInstalled(0), "Service should now be removed"
class ServiceTestCase(unittest.TestCase):
def checkInstallation(self):
"""Test service install/start/stop/remove"""
serviceInstall()
serviceStart()
serviceStop()
serviceRemove()
# After removing the service, the COM object should not exist
self.assertRaises(pywintypes.com_error, win32com.client.Dispatch, COMNAME)
class ComTestCase(unittest.TestCase):
"""Test the COM object in the service."""
# N.B. Because it takes a comparatively long time to install and remove the service
# the setUp installs it, but the tearDown leaves it installed.
# Individual tests start and stop the service as required.
def setUp(self):
if not isServiceInstalled():
OUT = "\nService installed\n"
commandWithOutput("service.py install", OUT)
assert isServiceInstalled(1), "Service should now be installed"
def checkServiceAutoStarts(self):
# The COM object isn't registered until we start the service,
# so make sure service has been started at least once
serviceStart()
serviceStop()
assert serviceState()==SERVICE_STOPPED
obj = win32com.client.Dispatch(COMNAME)
assert serviceState()==SERVICE_RUNNING
def checkComMethods(self):
serviceStop()
serviceStart()
obj = win32com.client.Dispatch(COMNAME)
assert obj.Hello("World")=="Hello World"
assert obj.Hello("Test2")=="Hello Test2"
assert obj.globalCalls()==2
assert obj.noCalls==2
del obj
obj = win32com.client.Dispatch(COMNAME)
assert obj.Hello("World")=="Hello World"
assert obj.globalCalls()==3
assert obj.noCalls==1
del obj
serviceStop()
serviceStart()
obj = win32com.client.Dispatch(COMNAME)
assert obj.Hello("World")=="Hello World"
assert obj.globalCalls()==1
assert obj.noCalls==1
def suite():
suite = unittest.makeSuite(ServiceTestCase, 'check')
suite.addTest(unittest.makeSuite(ComTestCase, 'check'))
return suite
allTests = suite()
if __name__=='__main__':
if len(sys.argv) > 1 and sys.argv[1] == 'gui':
import unittestgui
unittestgui.main("servicetest.allTests")
else:
verbose = 1
if '-v' in sys.argv or '--verbose' in sys.argv:
verbose += 1
runner = unittest.TextTestRunner(verbosity=verbose)
runner.run(allTests)
---- end servicetest.py ----
---- begin comobject.py ----
import sys
import pythoncom, win32api, time, win32con
globalCalls = 0
class HelloWorld:
_reg_clsctx_ = pythoncom.CLSCTX_LOCAL_SERVER
_reg_class_spec_ = "comobject.HelloWorld"
# Don't copy this: generate a new guid with pythoncom.CreateGuid()
_reg_clsid_ = "{8196223B-4BC8-4B91-A9E8-3FFA0A72FD5A}"
_reg_desc_ = "Python Test COM Server"
_reg_progid_ = "Python.TestServer"
_public_methods_ = ['Hello', 'globalCalls']
_public_attrs_ = ['softspace', 'noCalls']
_readonly_attrs_ = ['noCalls']
def __init__(self):
self.softspace = 1
self.noCalls = 0
def globalCalls(self):
return globalCalls
def Hello(self, who):
global globalCalls
self.noCalls = self.noCalls + 1
globalCalls += 1
# insert "softspace" number of spaces
msg = "Hello" + " " * self.softspace + str(who)
service.LOG("Message: %s" % msg)
return msg
def registerCOMservice(servicename, classes=HelloWorld):
try:
import win32com.server.register
register = win32com.server.register
register.RegisterClasses(classes)
# Now complete the registration for a service.
clsid = classes._reg_clsid_
keyNameRoot = "CLSID\\%s" % str(clsid)
register._set_subkeys(keyNameRoot, { 'AppId': str(clsid)} )
# The serviceparameter gets passed to us when started for COM.
Appidkeys = { "LocalService": servicename, "ServiceParameters": "LocalServiceStartup"}
register._set_subkeys("AppID\\%s" % clsid, Appidkeys)
except:
service.LOG("registerCOMservice", sys.exc_info())
def unregisterCOMservice(classes=HelloWorld):
import win32com.server.register
win32com.server.register.UnregisterClasses(classes)
# So that we can get at the logger
# This has to come after the class definition of the COM object.
import service
---- end comobject.py ----
---- begin service.py ----
# Very small service written in Python
import win32serviceutil
import win32service
import win32event, win32api
import time, sys, os
import comobject
import pythoncom, win32api, time, win32con
try:
p = os.path
LOGFILE = p.join(p.dirname(p.abspath(__file__)), "logfile.txt")
except NameError:
LOGFILE = "logfile.txt"
# This is the class we wish to make available via COM.
COMOBJECT = comobject.HelloWorld
def log_time():
"""Return a simple time string without spaces suitable for logging
"""
return ("%4.4d-%2.2d-%2.2dT%2.2d:%2.2d:%2.2d"
% time.gmtime(time.time())[:6])
class Logger:
def __init__(self, filename=LOGFILE):
self.logfile = open(filename, "w+t")
sys.stderr = sys.stdout = self.logfile
import os
print os.getcwd()
def log(self, summary, error=None):
time = log_time()
self.logfile.write("%s: %s\n" % (time, summary))
if error:
try:
self.logfile.write(format_exception(
error[0], error[1], error[2],
trailer='\n', limit=100))
except:
self.logfile.write("%s: %s\n" % error[:2])
self.logfile.flush()
def __call__(self, summary, error=None):
self.log(summary, error)
class MyService(win32serviceutil.ServiceFramework):
_svc_name_ = "PythonCOMService"
_svc_display_name_ = "Python COM service"
def __init__(self, *args):
try:
self.log = Logger()
global LOG
LOG = self.log
self.log("Service starting, args=%s" % args)
win32serviceutil.ServiceFramework.__init__(self, *args)
self.ReportServiceStatus(win32service.SERVICE_START_PENDING)
# Create an event which we will use to wait on.
# The service stop request will set this event.
self.hWaitStop = win32event.CreateEvent(None, 0, 0, None)
#comobject.registerCOMservice(self._svc_name_, self.log, COMOBJECT)
self.classes = [COMOBJECT._reg_clsid_]
except:
self.log("Exception during service start", error=sys.exc_info())
def SvcStop(self):
try:
# Tell the SCM we are starting the stop process.
self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING)
win32event.SetEvent(self.hWaitStop)
except:
self.log("SvcStop()", sys.exc_info())
# Stop the COM thread
#win32api.PostThreadMessage(self.threadId, win32con.WM_QUIT, 0, 0);
def SvcDoRun(self):
# Runs a COM loop
self.log("Running")
try:
self.ReportServiceStatus(win32service.SERVICE_START_PENDING)
sys.coinit_flags = 2
from win32com.server import factory
self.threadId = win32api.GetCurrentThreadId()
pythoncom.CoInitialize()
infos = factory.RegisterClassFactories(self.classes)
pythoncom.CoResumeClassObjects()
self.ReportServiceStatus(win32service.SERVICE_RUNNING)
while 1:
res = win32event.MsgWaitForMultipleObjects((self.hWaitStop,), 0, 2000, win32event.QS_ALLEVENTS)
if res == win32event.WAIT_OBJECT_0:
break # Must be terminating
pythoncom.PumpWaitingMessages()
factory.RevokeClassFactories(infos)
pythoncom.CoUninitialize()
except:
self.log("SvcDoRun()", sys.exc_info())
self.log("Service stopped")
self.ReportServiceStatus(win32service.SERVICE_STOPPED)
if __name__=='__main__':
# If we are removing the service, also remove the COM registration.
if "remove" in sys.argv:
comobject.unregisterCOMservice(COMOBJECT)
elif 'install' in sys.argv:
comobject.registerCOMservice(MyService._svc_name_, COMOBJECT)
win32serviceutil.HandleCommandLine(MyService)
---- end service.py ----
--
Duncan Booth duncan at rcp.co.uk
int month(char *p){return(124864/((p[0]+p[1]-p[2]&0x1f)+1)%12)["\5\x8\3"
"\6\7\xb\1\x9\xa\2\0\4"];} // Who said my code was obscure?
More information about the Python-list
mailing list