Singleton and Borg Design Pattern
The Singleton Design Pattern is quite popular.
As usual Python provides an easy and even better solution for most situations the Singleton seems appropriate. But first look at our options:
The classic Singleton pattern implemented in Python looks like this:
class Singleton(object):
def __new__(cls, *p, **k):
if not '_the_instance' in cls.__dict__:
cls._the_instance = object.__new__(cls)
return cls._the_instance
The Borg Design Pattern Borg Design Pattern shares just one state. I’ve found it at ASPN (Alex Martelli).
The main idea is to just share the state by replacing the default instance dictionary with a shared one:
class Borg(object):
_state = {}
def __new__(cls, *p, **k):
self = object.__new__(cls, *p, **k)
self.__dict__ = cls._state
return self
The best solution I found is to just use the Python mechanics: A Module is a singleton, why not just use it?
class _SingleInstance(object):
def __init__(self):
self.foo = 23
SingleInstance = _SingleInstance
del _SingleInstance
A good discussion about those pattern (and some others) can be found here: http://www.aleax.it/Python/5ep.html
The Borg design pattern can be found at ASPN: http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/66531
posted at: 13:05 | path: /python | permanent link to this entry
Using PyBlosxom
Searching for an extendable but simple weblog system, I finally found PyBlosxom.
This one is really simple and easy to extend. As a prove of concept I wrote the mod_python hander to integrate the blog in Apache 2. There is an old code snippled for this purpose somewhere out there. But this one won't work with recend software.
#!/usr/bin/python
"""
Pyblosxom mod_python handler
Based on ideas from
* Wari Wahab <wari at wari.per.sg>
http://roughingit.subtlehints.net/pyblosxom/2003/Jun/02/
* pyblosxom.cgi 1.3.2
"""
from mod_python import apache
import sys
sys.path.append('/etc/pyblosxom')
from config import py as cfg
if cfg.has_key("codebase"):
sys.path.insert(0, cfg["codebase"])
from Pyblosxom.pyblosxom import PyBlosxom
__author__ = 'Andreas Stricker <rapidmax@gmx.net>'
__version__ = "1.3.2"
__copyright__ = "Copyright (c) 2006 Andreas Stricker"
__license__ = "Python"
def handler(req):
req.content_type = 'text/html'
req.add_common_vars()
env = {}
for k, v in req.subprocess_env.items():
env[k] = v
env['wsgi.input'] = req
env['wsgi.errors'] = sys.stderr
# setup url_scheme for static rendering
if cfg.has_key('base_url'):
if cfg['base_url'].find('://') > 0:
env['wsgi.url_scheme'] = cfg['base_url'][:cfg['base_url'].find("://")]
else:
env['wsgi.url_scheme'] = "http"
if cfg['base_url'].startswith('/'):
cfg['base_url'] = "http://" + req.hostname + cfg['base_url']
else:
env['wsgi.url_scheme'] = "http"
cfg['base_url'] = "/blog/blog.py"
p = PyBlosxom(cfg, env)
p.run()
response = p.getResponse()
# detect if we have to return "404 Not found" status
# This case is handled as suggested in mod_python documenation
if response.status.startswith('404'):
req.status = apache.HTTP_NOT_FOUND
# set headers
for k, v in response.headers.items():
if k.lower() == 'content-type':
req.content_type = v
req.headers_out[k] = v
response.sendBody(req)
return apache.OK
To make it work, you must register this script as handler for a specific directory:
DirectoryIndex blog.py AddHandler mod_python .py PythonHandler blog #PythonDebug On
posted at: 14:58 | path: /python | permanent link to this entry
Building Finite State Machines (FSM) with Python
A problem that many times occured to me. I tried to implement a finite state machine, but there was no convenient framework to do this. Of corse there are powerful RAD Tools (e.g. for SDL), but I didn’t wanted this dependency. The framework should be simple, clean and easy to use but still powerfull enough to describe the problem. I know there is alway a switch statement FSM. This is good enough for simple state machines.
I ended up by implementing something for myself. The first try was a C++ module (see SIPSec ). two years later I played a bit with a Implementation in Python. This one worked, but didn’t satisfied me enough. It was still to complex.
Once again nearly a year later I had the Idea. I’ll provide the full documented source later. First I developp the ideas mixed with concrete code.
For a full state machine we need:
- States
- Transition between states
- Events to initiate a transition
- Operations to execute during a Transition
Further we should have (but not necessary need for a FSM by definition):
- Conditionals (depending on the value of data transported in an event)
- Operations executed on enter and leave event of a State
Let’s start at the end: We like to implement an example FSM looking like that:
+-----------------+ | +-+e1 | V e1,e3 V | e2 |e1 (A) ---> (B) ---> (C) <-+ ^ ^ |e2 | | | V | | | /x\ | | | /0_1\ | | | /0 1 2\ | | | | | | | | +------+ | +--+ +-----------------+
The first thing we need are States. We can define those like this:
a = State('A')
b = State('B')
c = State('C')
Fine, but how does a state look like? We have one methode process() that takes an event and decide if and which transition it has to choose. And then we have some methods needed to construct the state machine, adding transitions and operations. Have a look at the code
class State(object): ON_ENTER = ‘on_enter’ ON_LEAVE = ‘on_leave’ def __init__(self, name): self.name = name self.on_enter = [] # callbacks self.on_leave = [] # callbacks self.events = {} # events -> transition def __repr__(self): return self.name def add_operation(self, when, operation): getattr(self, when).append(operation) return self def add_event(self, event, transition): self.events[event] = transition return self def enter(self): for op in self.on_enter: op(self) def leave(self): for op in self.on_leave: op(self) def process(self, event): if not self.events.has_key(event): return self # sink self.leave() return self.events[event].process(event)
For the building phase the add_event() method is important: It add a transition initiated by an event to the internal lookup-table (dict). If an event occurs it is submitted to the process() method. This function look up the event in the table and execute its transition process() method if an event exists or return itself without furter processing if not. The enter() and leave() functions should be clear.
We talked about a transition, but never saw one. May I introduce them? A transition is the thing between two states on time they changes. This is if the old state is left and before the new one comes. A Transition defines which state comes next. And it also provides a hook for operations to execute on this transition. Likewise to the state the transition consist of methods to build the machine and of methodes used at execution time. When we instantiate a transition we define the end state we enter.
class Transition(object):
def __init__(self, to_state):
self.to_state = to_state
self.operations = []
def add_operation(self, op):
self.operations.append(op)
return self
def process(self, event):
for op in self.operations:
op(event)
self.to_state.enter()
return self.to_state
The process() method returns the state we end in. To connect two states, we do something like this:
b.add_event(e2, Transition(c).add_operation(Printer("Transition b -> c")))
This adds a transition to state B on event e thats execute the callable Printer() on transition.
Finally we are missing the events. Those aren’t nothing special. We could have choosen to differentiate on class or just choosen a single object (dictionary for example). But I think it’s best to have a single Event class and differentiate based on a ID given on construction time. The ID should be a string (but could also be an integer). We are still able to subclass the Event if we need not only to set atributes but also to give a behaviour to an event. So look at the implementation: It’s somewhat pythonic:
class Event(object):
def __init__(self, event_id):
self.event_id = event_id
def __hash__(self):
return hash(self.event_id)
def __cmp__(self, other):
if isinstance(other, Event):
return cmp(self.event_id, other.event_id)
else:
return cmp(self.event_id, other)
def __repr__(self):
return "<Event %s>" % `self.event_id`
Finally we define some events for our example:
e1 = Event('e1')
e2 = Event('e2')
e3 = Event('e3')
This is nearly all we need for our example FSM. But what we need an entry point, an interface to access the whole thing. I’ll call them FSM because it represent an FSM (even if it’s deadly simple!):
class FSM(object):
def __init__(self, start):
self.state = self.start = start
def process(self, event):
self.state = self.state.process(event)
If you look at our example you seed that there is still on missing brick: The conditional. I implemented this one as generic as possible: First it subclass our transition and second it ends up in transitions (there must be at least one). The Conditional has finite ends (I call them paths) and a desition to choose one of those paths. The paths are simple implemented by yet another lookup table (dict) and the decition is just a callable object (function, method, lambda or callable class). Once again we don’t have to subclass the Conditional class for every single task.
class Conditional(Transition): def __init__(self, decision, paths): Transition.__init__(self, None) self.decision = decision self.paths = paths def process(self, event): for op in self.operations: op(event) return self.paths[self.decision(event)].process(event)
Simple, isn’t it?
Let’s do some action: Here is the code to implement the FSM as shown above:
class Printer(object): def __init__(self, txt): self.txt = txt def __call__(self, *args): print self.txt, str(args)a = State(‘A’)
b = State(‘B’)
c = State(‘C’)
e1 = Event(‘e1’)
e2 = Event(‘e2’)
e3 = Event(‘e3’)t = Transition(b).add_operation(Printer(“Transition a -> b”))
a.add_event(e1, t).add_event(e3, t)
b.add_event(e1, Transition(b).add_operation(Printer(“Transition b -> b”))\ .add_operation(Printer(“2nd Operation”)))
b.add_event(e2, Transition©.add_operation(Printer(“Transition b -> c”)))
c.add_event(e1, Transition(a).add_operation(Printer(“Transition c -> a”)))
c.add_event(e2, Conditional(lambda event: event.x, { 0 : Transition(b).add_operation(Printer(“Transition c -> b”)), 1 : Transition(a).add_operation(Printer(“Transition c -> a”)), 2 : Transition©.add_operation(Printer(“Transition c -> c”)),
}).add_operation(Printer(“Decision (x = ?) -> a,b,c”)))a.add_operation(‘on_enter’, Printer(“Enter State A”))
a.add_operation(‘on_leave’, Printer(“Leave State A”))
b.add_operation(‘on_enter’, Printer(“Enter State B”))
b.add_operation(‘on_leave’, Printer(“Leave State B”))
c.add_operation(‘on_enter’, Printer(“Enter State C”))
c.add_operation(‘on_leave’, Printer(“Leave State C”))fsm = FSM
That’s it. Next we jump a bit around just to see if it works:
fsm.process(e1) fsm.process(e1) fsm.process(e2) fsm.process(e1) fsm.process(e3) fsm.process(e2) e2.x = 2 fsm.process(e2) e2.x = 0 fsm.process(e2) fsm.process(e2) fsm.process(e2)
The output should look like this:
Leave State A (A,) Transition a -> b (<Event 'e1'>,) Enter State B (B,) Leave State B (B,) Transition b -> b (<Event 'e1'>,) 2nd Operation (<Event 'e1'>,) Enter State B (B,) Leave State B (B,) Transition b -> c (<Event 'e2'>,) Enter State C (C,) Leave State C (C,) Transition c -> a (<Event 'e1'>,) Enter State A (A,) Leave State A (A,) Transition a -> b (<Event 'e3'>,) Enter State B (B,) Leave State B (B,) Transition b -> c (<Event 'e2'>,) Enter State C (C,) Leave State C (C,) Decision (x = ?) -> a,b,c (<Event 'e2'>,) Transition c -> c (<Event 'e2'>,) Enter State C (C,) Leave State C (C,) Decision (x = ?) -> a,b,c (<Event 'e2'>,) Transition c -> b (<Event 'e2'>,) Enter State B (B,) Leave State B (B,) Transition b -> c (<Event 'e2'>,) Enter State C (C,) Leave State C (C,) Decision (x = ?) -> a,b,c (<Event 'e2'>,) Transition c -> a (<Event 'e2'>,) Enter State A (A,)
posted at: 23:48 | path: /python | permanent link to this entry
PyBlosxom generic Wiki markup
It’s nice to format the blog entries in HTML. Of course HTML is somewhat complicated. I like Wiki markup: It is straightforward and simple.
To add Wiki markup support to PyBlosxom, there is a plugin in entryparsers subdirectory of the contrib package. Not bad, but I’m missing a code block option. Once again I was able to enhance the source code.
Apropos: This entry is formated with the enhanced genericwiki parser. Here the prove:
“”“
Generic wiki markup PreFormatter 2002-11-18, for pyblosxom
CHANGE wikibaseurl to point to your wiki, & wikinamepattern to yours
Bug reports, comments, presents, etc. to John Abbe at johnca@ourpla.net
ToDo: Lists; code; InterWiki links; other wikinamepatternsYou can configure this as your default preformatter by configuring it in your
py[‘parser’] = ‘genericwiki’
L{config} file as follows::or in your blosxom entries, place a C{#parser wiki} line after the title of
My Little Blog Entry #parser genericwiki This is a text in ‘’‘wiki’‘’ format
your blog::This preformatter also supports WikiWirds, you need to point out where your
py[‘genericwiki_baseurl’] = ‘http://www.google.com/search?q=’
Wiki site is. This is configured with a new variable in your config.py file,
‘genericwiki_baseurl’::The above example would expand ‘WikiWord’ to
http://www.google.com/search?q=WikiWordPermission is hereby granted, free of charge, to any person
obtaining a copy of this software and associated documentation
files (the “Software”), to deal in the Software without restriction,
including without limitation the rights to use, copy, modify,
merge, publish, distribute, sublicense, and/or sell copies of the
Software, and to permit persons to whom the Software is furnished
to do so, subject to the following conditions:The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.Copyright 2004, 2005 John Abbe
“”“
author = ‘John Abbe’
version = “$Id: genericwiki.py,v 1.1.2.1 2005/06/21 18:22:33 willhelm Exp $”
PREFORMATTER_ID = ‘genericwiki’
import re
- caching compiled regular expressions
re_cache = Nonedef cb_preformat(args): “”“ Preformat callback chain looks for this.
param args: a dict with 'parser' string and a list 'story'type args: dict “”“ if args[‘parser’] == PREFORMATTER_ID: config = args[‘request’].getConfiguration() baseurl = config.get(‘genericwiki_baseurl’, None) return parse(‘’.join(args[‘story’]), baseurl)def _build_cache(): “”“ Compile regular expression pattern and store them in global cache “”“ global re_cache # build the cache first time called (speedup later calls!) re_cache = {}
# WikiName pattern used in your wiki wikinamepattern = r’\b(([A-Z]+[a-z]+){2,})\b’ # original mailurlpattern = r’mailto\:[\”\-\_\.\w]+\@[\-\_\.\w]+\w’ newsurlpattern = r’news\:(?:\w+\.){1,}\w+’ fileurlpattern = r’(?:http|https|file|ftp):[/-_.\w-]+[\/\w][?&+=%\w/-_.#]*’ # Turn ‘[xxx:address label]’ into labeled link re_cache[‘bracket_links’] = re.compile(r’\[(’ + fileurlpattern + ‘|’ + mailurlpattern + ‘|’ + newsurlpattern + ‘)\s+(.+?)\]’) # Convert naked URLs into links — skip ones with a “ before re_cache[‘nacked_url’] = re.compile(r’(?def _parse_markup(text, wikibaseurl): “”“ Parse wiki markup for textparam text: Text for conversiontype text: string “”“ if not len(text.strip()): return “” # Turn ‘[xxx:address label]’ into labeled link text = re_cache[‘bracket_links’].sub(r’\2’, text) # Convert naked URLs into links — skip ones with a “ before text = re_cache[‘nacked_url’].sub(r’\1’, text) # Convert WikiNames into links if wikibaseurl: text = re_cache[‘wiki_names’].sub(‘\1’, text) # ‘’ for emphasis, ‘’‘ for strong, —— for a horizontal rule text = re_cache[‘strong’].sub(r”\1”, text) text = re_cache[‘emphasis’].sub(r”\1”, text) text = re_cache[‘ruler’].sub(”
”, text) # Convert two or more newlines intotext = re_cache[‘para’].sub(r’
\n’, text) return “
” + text + “
”def _parse_code(text): if text.strip().startswith(‘#!python’): keywords = “(and|del|for|is|raise|assert|elif|from|lambda|return|break”\ “|else|global|not|try|class|except|if|or|while|continue”\ “|exec|import|pass|yield|def|finally|in|print)” text = re.sub(keywords, r’\1’, text) return text
def parse(text, wikibaseurl): “”“ The main workhorse that convert wiki text into html markup
param text: Text for conversiontype text: string @return: string “”“ global re_cache if not re_cache: _build_cache() # replace entities text = text.replace(‘&’, ‘&’).replace(‘<’, ‘<’).replace(‘>’, ‘>’) # convert preformated text and wiki markup text fragments = [ x.split(‘} }}’, 1) for x in text.split(‘{ {{‘) ] result = [] for fragment in fragments: if len(fragment) > 1: result.append(“<” + “pre” + “>” + _parse_code(fragment0) + “”) result.append(_parse_markup(fragment1, wikibaseurl)) else: result.append(_parse_markup(fragment0, wikibaseurl)) return “”.join(result)if name == ‘main’: text = “”“This is a test To test the wiki
[http://roughingit.subtlehints.net/pyblosxom?blah=duh#spam A link] news:roughingit.subtlehints.net/pyblosxom/ – no, ‘‘I’‘ do ‘’‘not’‘’ have a news server. mailto:wari@example should go link to an email. WikiWiki is a wiki Keyword A text for preformated text: {{{ #!python import this print “Yes! don’t replace ‘‘source’‘ ‘’‘code’‘’ here: http://don.t.do.this.ch/” }}} That’s it, folks! “”“ print parse(text, ‘http://wiki.subtlehints.net/moin/’)
Note: You should remove the space in curly brackets.
posted at: 23:36 | path: /python | permanent link to this entry
Adding new messages with e-mail
This new python module allows to write new messages in your mail program and send it to the blog
Following Items are needed to successful blog with mail:
- Subject for the title
- Passphrase to authenticate
- Path with category and filename (no trailing slash!)
- A empty line (as with RFC822 header/body
- The blog body, with markup
This script will do the work. It not finished yet, there are some problems to solve.
#!/usr/bin/python
"""
Script to check a IMAP folder for specific blog mails to
upload to a weblog via xmlrpc
"""
import getpass
import xmlrpclib
import imaplib
import email.Parser
passphrase = "A good passphrase"
USERNAME = 'yourname'
class IMAPError(Exception): pass
class BlogEntry(object):
def __init__(self, data=None):
self.title = None
self.body = None
self.path = None
self.valid = False
if data:
self.parse(data)
def __repr__(self): return self.title
def parse(self, data):
parser = email.Parser.Parser()
em = parser.parsestr(data)
if not em.has_key('Subject'):
return
content = em.get_payload()
if em.is_multipart():
content = content[0] # only use first part
content = parser.parsestr(content) # body is a pseudo RFC822 text
if not content.get('Passphrase') or not content.get('Path'):
return
#if not em.get('Passphrase') == passphrase:
# return
self.title = em.get('Subject')
self.path = content.get('Path')
if not self.path.startswith('/'):
self.path = '/' + self.path
if not self.path.endswith('/'):
self.path += '/'
self.body = content.get_payload()
self.valid = True
def get_blog_mails(server, mailbox, username, password):
m = imaplib.IMAP4(server)
status, data = m.login(username, password)
if not status.startswith('OK'):
raise IMAPError, status
status, data = m.select(mailbox)
if not status.startswith('OK'):
raise IMAPError, status
status, data = m.search(None, 'ALL')
if not status.startswith('OK'):
raise IMAPError, status
mail_id_list = data[0].split()
result = []
for mailid in mail_id_list:
status, data = m.fetch(mailid, '(RFC822)')
result.append(data[0][1])
# TODO ast: mark mail as read
m.logout()
return result
class BloggerAPI(object):
APPKEY = "070F5B78CF8DF3450B732A53546CCFC6636A4547"
def __init__(self, url, username, password):
self.server = xmlrpclib.Server(url)
self.username = username
self.password = password
def getUsersBlogs(self):
return self.server.blogger.getUsersBlogs(self.APPKEY,
self.username,
self.password)
def newPost(self, blogid, content, publish=1):
self.server.blogger.newPost(self.APPKEY,
blogid,
self.username,
self.password,
content,
publish)
def getRecentPosts(self, blogid, numberOfPosts=5):
return self.server.blogger.getRecentPosts(self.APPKEY,
blogid,
self.username,
self.password,
numberOfPosts)
def getUserInfo(self):
return self.server.blogger.getUserInfo(self.APPKEY,
self.username,
self.password)
if __name__ == '__main__':
password_mail = getpass.getpass("Your mailbox password: ")
password_blog = getpass.getpass("Your blog password: ")
blogger = BloggerAPI('http://sirius/blog/blog.py/RPC',
USERNAME, password_blog)
for mail in get_blog_mails("sirius.hogwarts.local",
"INBOX.projects.weblog",
USERNAME, password_mail):
blog = BlogEntry(mail)
print blog.title, blog.path
try:
blogger.newPost(blog.path, blog.title + "\n" + blog.body)
except xmlrpclib.Fault, e:
print e
posted at: 01:53 | path: /python | permanent link to this entry
