Skip to content

Instantly share code, notes, and snippets.

@jsbueno
Created March 28, 2023 18:10
Show Gist options
  • Save jsbueno/f462e23c24a6fb8ace01c88d9bcf6f58 to your computer and use it in GitHub Desktop.
Save jsbueno/f462e23c24a6fb8ace01c88d9bcf6f58 to your computer and use it in GitHub Desktop.
"bicycle repair man"

"Bicycle Repair Man" script showing a "zope-esque" prototype in less than 200 LoC using multiple inheritance and mixins.

This prototype was presented in Python Brasil 2011 in Sao Paulo, explaining at once a bit about what Zope does (did), and showng the capabilities of multiple inheritance.

(and likely not touched since).

Unfortunatelly I think the video for this lecture had not being made, or not survived - the other lectures in the same PyCon are here: https://www.youtube.com/watch?v=MDq4KecP3Bg&list=PLDC3uVLxaEQ3cPuehu65qU-QSNj6etm1q&ab_channel=pythonbrasil

# -*- coding: utf-8 -*-
import wsgiref
from wsgiref.simple_server import make_server
from array import array
import itertools
import cPickle as pickle
import sys
mainfile = "data.fs"
MAXOBJ = 100
OFF_ITEM_SIZE = array("l").itemsize
"""
Persistence object file description:
a single file - the first OFF_ITEM_SIZE objects describe the offset where the
root object is pickled too.
Each object is followed by MAXOBJ * OFF_ITEM_SIZE bytes, indicating the
offsets of contained objects in the file
"""
class Tree(object):
"""
Basic mapping class supporting hierarchical structure
and ordered contents.
"""
folderish = True
def __init__(self, name, parent):
self.children = {}
self.child_count = 0
self.parent = parent
self.name = name
if parent is not None:
self.parent[name] = self
def __getitem__(self, name):
return self.children[name]["obj"]
def __setitem__(self, name, obj):
if not name in self.children:
self.children[name] = {}
self.children[name]["order"] = self.child_count
self.child_count += 1
self.children[name]["obj"] = obj
obj.name = name
#TODO: Add support to other methods on dictionary model later
def get_url(self):
path = "/" + self.name
parent = self.parent
while parent is not None:
path = "/" + parent.name + path
parent = parent.parent
return path
class Traverse(Tree):
"""
Implements traversal attribute search:
retrieves attributes of higer elements on the
on the tree structure if there is no child or
attribute with the required name
"""
def __getattr__(self, attr):
if attr in self.children:
return self[attr]
if self.parent is not None:
return getattr(self.parent, attr)
raise AttributeError ("%s does not exist" % attr)
class Persist(Tree):
def __setitem__(self, name, obj):
self.dirty = True
return super(Persist, self).__setitem__(name, obj)
def __getstate__(self):
dct = self.__dict__.copy()
dct["children"] = self.children.copy()
for key in dct["children"].keys():
dct["children"][key] = dct["children"][key].copy()
dct["children"][key]["obj"] = None
dct["parent"] = None
dct["dirty"] = False
return dct
def _save(self):
# Move to end of file:
data.seek(0, 2)
self.offset = data.tell()
pickle.dump(self, data, -1)
self.contents_offset = data.tell()
offset_data = []
for name, obj in sorted(self.children.items(), key = lambda item: item[1]["order"]):
offset_data.append(getattr("obj", "offset", -1))
offset_data.extend(itertools.repeat(-1, MAXOBJ - len(self.children)))
offset_data = array("l", offset_data)
offset_data.tofile(data)
#Update self position in parent's content index:
if self.parent is not None:
index_in_parent = self.parent.children[self.name]["order"]
position = self.parent.contents_offset + index_in_parent * OFF_ITEM_SIZE
else:
position = 0
data.seek(position, 0)
pos = array("l", [self.offset])
pos.tofile(data)
self.dirty = False
def commit(self):
if self.parent and hasattr(self.parent, "dirty") and self.parent.dirty:
self.parent.commit()
self._save()
def __setstate__(self, state):
#called by pickle on restoring -
self.__dict__.update(state)
stream_offset = data.tell()
#walks over the end of pickle data. For protocol "2" it
# is a single "." (chr(46)) byte
data.seek(1,1)
offsets = array("l")
offsets.fromfile(data, MAXOBJ)
for name, entry in self.children.items():
entry["offset"] = offsets[entry["order"]]
data.seek(stream_offset)
def __getitem__(self, name):
if name in self.children and self.children[name]["obj"] == None:
# Load object from disk
data.seek(self.children[name]["offset"], 0)
self.children[name]["obj"] = pickle.load(data)
self.children[name]["obj"].parent = self
return super(Persist, self).__getitem__(name)
class Display(object):
view_template = u"""<h1>%(title)s</h1>"""
def view_repr(self, environ, start_response):
start_response('200 OK', [('Content-type', 'text/html; charset=utf-8')])
return [(self.view_template % self.todict()).encode("utf-8")]
__call__ = view_repr
def todict(self):
dct = {}
for attr_name in dir(self):
if attr_name.startswith("_"):
continue
attribute = getattr(self, attr_name)
if isinstance(attribute, unicode):
dct[attr_name] = attribute
return dct
def __repr__(self):
return "Bicicle Repair Man Web Object"
@property
def title(self):
return u"Bicicle Repair Man Web Object: %s" % self.name.decode("utf-8")
class BicicleRepairMan(Traverse, Persist, Display):
pass
class Folder(BicicleRepairMan):
view_template = u"""
<h1>%(title)s</h1>
<ul>
%(children_list)s
</ul>
"""
@property
def children_list(self):
templ = u"""<li><a href="%s">%s</li>"""
html = ""
for name in self.children:
html += templ %(self.get_url() + "/" + name, name)
return html
def fetch_offset(file):
data = array("l")
data.fromfile(file,1)
return data[0]
try:
data = open("data.fs", "rb+")
root_offset = fetch_offset(data)
data.seek(root_offset)
root = pickle.load(data)
except (IOError,EOFError) :
data = open("data.fs", "wb+")
array("l", [OFF_ITEM_SIZE]).tofile(data)
root = BicicleRepairMan("root", None)
root.commit()
data.flush()
def dispatcher(environ, start_response):
path_info = environ["PATH_INFO"].split("/")
obj = root
for name in path_info:
if not name:
continue
try:
obj = getattr(obj, name)
except AttributeError:
start_response('404 NOT FOUND', [('Content-type', 'text/html; charset=utf-8')])
return (["<h1>404 NOT FOUND</h1>"])
return obj(environ, start_response)
def init():
r = root
s = BicicleRepairMan("spawn", r)
s2 = BicicleRepairMan("other", r)
s3 = BicicleRepairMan("spawnned", s)
s3.commit()
s2.commit()
s = Folder("folder", r)
for x in range(5):
s1 = Folder("child_%d" %x, s)
s1.commit()
for x in range(3):
s2 = Folder("grandchild_%d" %x, s1)
s2.commit()
if __name__ == "__main__":
if len(sys.argv) == 2 and sys.argv[1] == "reset":
init()
server = make_server("127.0.0.1", 8000, dispatcher)
server.serve_forever()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment