/usr/share/pyshared/axiom/scripts/axiomatic.py is in python-axiom 0.7.1-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 | # -*- test-case-name: axiomatic.test.test_axiomatic -*-
from zope.interface import alsoProvides, noLongerProvides
import os
import sys
import glob
import errno
import signal
from twisted.plugin import IPlugin, getPlugins
from twisted.python import usage
from twisted.python.runtime import platform
from twisted.scripts import twistd
from axiom.iaxiom import IAxiomaticCommand
class AxiomaticSubCommandMixin(object):
store = property(lambda self: self.parent.getStore())
def decodeCommandLine(self, cmdline):
"""Turn a byte string from the command line into a unicode string.
"""
codec = getattr(sys.stdin, 'encoding', None) or sys.getdefaultencoding()
return unicode(cmdline, codec)
class _AxiomaticCommandMeta(type):
"""
Metaclass for L{AxiomaticCommand}.
This serves to make subclasses provide L{IPlugin} and L{IAxiomaticCommand}.
"""
def __new__(cls, name, bases, attrs):
newcls = type.__new__(cls, name, bases, attrs)
alsoProvides(newcls, IPlugin, IAxiomaticCommand)
return newcls
class AxiomaticSubCommand(usage.Options, AxiomaticSubCommandMixin):
"""
L{twisted.python.usage.Options} subclass for Axiomatic sub commands.
"""
class AxiomaticCommand(usage.Options, AxiomaticSubCommandMixin):
"""
L{twisted.python.usage.Options} subclass for Axiomatic plugin commands.
Subclass this to have your class automatically provide the necessary
interfaces to be picked up by axiomatic.
"""
__metaclass__ = _AxiomaticCommandMeta
noLongerProvides(AxiomaticCommand, IPlugin)
noLongerProvides(AxiomaticCommand, IAxiomaticCommand)
class PIDMixin:
def _sendSignal(self, signal):
if platform.isWindows():
raise usage.UsageError("You can't send signals on Windows (XXX TODO)")
dbdir = self.parent.getStoreDirectory()
serverpid = int(file(os.path.join(dbdir, 'run', 'axiomatic.pid')).read())
os.kill(serverpid, signal)
return serverpid
def signalServer(self, signal):
try:
return self._sendSignal(signal)
except (OSError, IOError), e:
if e.errno in (errno.ENOENT, errno.ESRCH):
raise usage.UsageError('There is no server running from the Axiom database %r.' % (self.parent.getStoreDirectory(),))
else:
raise
class Stop(usage.Options, PIDMixin):
def postOptions(self):
self.signalServer(signal.SIGINT)
class Status(usage.Options, PIDMixin):
def postOptions(self):
dbdir = self.parent.getStoreDirectory()
serverpid = self.signalServer(0)
print 'A server is running from the Axiom database %r, PID %d.' % (dbdir, serverpid)
class Start(twistd.ServerOptions):
run = staticmethod(twistd.run)
def subCommands():
raise AttributeError()
subCommands = property(subCommands)
def getArguments(self, store, args):
run = store.dbdir.child("run")
logs = run.child("logs")
if "--logfile" not in args and "-l" not in args and "--nodaemon" not in args and "-n" not in args:
if not logs.exists():
logs.makedirs()
args.extend(["--logfile", logs.child("axiomatic.log").path])
if not platform.isWindows() and "--pidfile" not in args:
args.extend(["--pidfile", run.child("axiomatic.pid").path])
args.extend(["axiomatic-start", "--dbdir", store.dbdir.path])
return args
def parseOptions(self, args):
if "--help" in args:
self.opt_help()
else:
# If a reactor is being selected, it must be done before the store
# is opened, since that may execute arbitrary application code
# which may in turn install the default reactor.
if "--reactor" in args:
reactorIndex = args.index("--reactor")
shortName = args[reactorIndex + 1]
del args[reactorIndex:reactorIndex + 2]
self.opt_reactor(shortName)
sys.argv[1:] = self.getArguments(self.parent.getStore(), args)
self.run()
class Options(usage.Options):
def subCommands():
def get(self):
yield ('start', None, Start, 'Launch the given Axiom database')
if not platform.isWindows():
yield ('stop', None, Stop, 'Stop the server running from the given Axiom database')
yield ('status', None, Status, 'Report whether a server is running from the given Axiom database')
from axiom import plugins
for plg in getPlugins(IAxiomaticCommand, plugins):
try:
yield (plg.name, None, plg, plg.description)
except AttributeError:
raise RuntimeError("Maldefined plugin: %r" % (plg,))
return get,
subCommands = property(*subCommands())
optParameters = [
('dbdir', 'd', None, 'Path containing axiom database to configure/create'),
]
optFlags = [
('debug', 'b', 'Enable Axiom-level debug logging')]
store = None
def usedb(self, potentialdb):
yn = raw_input("Use database %r? (Y/n) " % (potentialdb,))
if yn.lower() in ('y', 'yes', ''):
self['dbdir'] = potentialdb
else:
raise usage.UsageError('Select another database with the -d option, then.')
def getStoreDirectory(self):
if self['dbdir'] is None:
possibilities = glob.glob('*.axiom')
if len(possibilities) > 1:
raise usage.UsageError(
"Multiple databases found here, please select one with "
"the -d option: %s" % (' '.join(possibilities),))
elif len(possibilities) == 1:
self.usedb(possibilities[0])
else:
self.usedb(self.subCommand + '.axiom')
return self['dbdir']
def getStore(self):
from axiom.store import Store
if self.store is None:
self.store = Store(self.getStoreDirectory(), debug=self['debug'])
return self.store
def postOptions(self):
if self.store is not None:
self.store.close()
def main(argv=None):
o = Options()
try:
o.parseOptions(argv)
except usage.UsageError, e:
raise SystemExit(str(e))
|