opendev/lodgeit
Code Issues Proposed changes

updated to sqlalchemy 0.4, moved and cleaned up the code

Browse Source
This commit is contained in:
entequak 2007-12-13 21:01:08 +01:00
parent 4b0df72c0c
commit 3399c3b322
11 changed files with 388 additions and 304 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

134
lodgeit/application.py
View File

@ -11,96 +11,13 @@
import os import os
import sqlalchemy import sqlalchemy
from datetime import datetime, timedelta from datetime import datetime, timedelta
from werkzeug.wrappers import BaseRequest, BaseResponse from werkzeug import SharedDataMiddleware, ClosingIterator
from werkzeug.utils import SharedDataMiddleware from lodgeit.utils import _local_manager, ctx, jinja_environment, \
from werkzeug.routing import NotFound, RequestRedirect Request, generate_user_hash, NotFound, RequestRedirect, redirect
from jinja import Environment, PackageLoader from lodgeit.database import metadata, db, Paste
from lodgeit.lib.antispam import AntiSpam
from lodgeit.urls import urlmap from lodgeit.urls import urlmap
from lodgeit.controllers import get_controller from lodgeit.controllers import get_controller
from lodgeit.database import metadata, generate_user_hash, Paste
from lodgeit.lib.antispam import AntiSpam
#: jinja environment for all the templates
jinja_environment = Environment(loader=PackageLoader('lodgeit', 'views',
use_memcache=False,
auto_reload=False
))
def datetimeformat():
"""
Helper filter for the template
"""
def wrapped(env, ctx, value):
return value.strftime('%Y-%m-%d %H:%M')
return wrapped
jinja_environment.filters['datetimeformat'] = datetimeformat
def render_template(req, template_name, **context):
"""
Render a template to a response. This automatically fetches
the list of new replies for the layout template. It also
adds the current request to the context. This is used for the
welcome message.
"""
if req.method == 'GET':
context['new_replies'] = Paste.fetch_replies(req)
context['request'] = req
t = jinja_environment.get_template(template_name)
return Response(t.render(context), mimetype='text/html; charset=utf-8')
def redirect(url, code=302):
"""
Redirect to somewhere. Returns a nice response object.
"""
return Response('Page Moved to %s' % url,
headers=[('Location', url),
('Content-Type', 'text/plain')],
status=code)
class Request(BaseRequest):
"""
Subclass of the `BaseRequest` object. automatically creates a new
`user_hash` and sets `first_visit` to `True` if it's a new user.
It also stores the engine and dbsession on it.
"""
charset = 'utf-8'
def __init__(self, environ, app):
self.app = app
self.engine = app.engine
self.dbsession = sqlalchemy.create_session(app.engine)
super(Request, self).__init__(environ)
# check the user hash. an empty cookie is considered
# begin a new session.
self.user_hash = ''
self.first_visit = False
if 'user_hash' in self.cookies:
self.user_hash = self.cookies['user_hash']
if not self.user_hash:
self.user_hash = generate_user_hash()
self.first_visit = True
class Response(BaseResponse):
"""
Subclass the response object for later extension.
"""
charset = 'utf-8'
class PageNotFound(NotFound):
"""
Internal exception used to tell the application to show the
error page.
"""
class LodgeIt(object): class LodgeIt(object):
@ -114,40 +31,53 @@ class LodgeIt(object):
self.engine = sqlalchemy.create_engine(dburi) self.engine = sqlalchemy.create_engine(dburi)
#: make sure all tables exist. #: make sure all tables exist.
metadata.create_all(self.engine) metadata.create_all(self.engine)
#: create a new antispam instance #: bind the application to the current context local
self.antispam = AntiSpam(self) self.bind_to_context()
#: create a new AntiSpam instance
self.antispam = AntiSpam()
def bind_to_context(self):
ctx.application = self
def __call__(self, environ, start_response): def __call__(self, environ, start_response):
""" """
Minimal WSGI application for request dispatching. Minimal WSGI application for request dispatching.
""" """
req = Request(environ, self) #: bind the application to the new context local
self.bind_to_context()
request = Request(environ)
request.bind_to_context()
urls = urlmap.bind_to_environ(environ) urls = urlmap.bind_to_environ(environ)
try: try:
endpoint, args = urls.match(req.path) endpoint, args = urls.match(request.path)
handler = get_controller(endpoint, req) handler = get_controller(endpoint)
resp = handler(**args) resp = handler(**args)
except NotFound: except NotFound:
handler = get_controller(self.not_found[0], req) handler = get_controller(self.not_found[0])
resp = handler(**self.not_found[1]) resp = handler(**self.not_found[1])
except RequestRedirect, err: except RequestRedirect, err:
resp = redirect(err.new_url) resp = redirect(err.new_url)
else: else:
if req.first_visit: if request.first_visit:
resp.set_cookie('user_hash', req.user_hash, resp.set_cookie('user_hash', request.user_hash,
expires=datetime.utcnow() + timedelta(days=31) expires=datetime.utcnow() + timedelta(days=31)
) )
# call the response as WSGI app
return resp(environ, start_response) return ClosingIterator(resp(environ, start_response),
[_local_manager.cleanup, db.session.remove])
def make_app(dburi): def make_app(dburi, debug=False, shell=False):
""" """
Apply the used middlewares and create the application. Apply the used middlewares and create the application.
""" """
static_path = os.path.join(os.path.dirname(__file__), 'static') static_path = os.path.join(os.path.dirname(__file__), 'static')
app = LodgeIt(dburi) app = LodgeIt(dburi)
app = SharedDataMiddleware(app, { if debug:
'/static': static_path app.engine.echo = True
}) if not shell:
# we don't need access to the shared data middleware in shell mode
app = SharedDataMiddleware(app, {
'/static': static_path
})
return app return app

13
lodgeit/controllers/__init__.py
View File

@ -11,18 +11,13 @@
class BaseController(object): class BaseController(object):
""" """
Base controller. add some stuff to the dict on instanciation Base controller class. This does nothing *yet* but
maybe is usefull later.
""" """
def __init__(self, req):
self.request = req
self.app = req.app
self.engine = req.engine
self.dbsession = req.dbsession
def get_controller(name):
def get_controller(name, req):
cname, hname = name.split('/') cname, hname = name.split('/')
module = __import__('lodgeit.controllers.' + cname, None, None, ['']) module = __import__('lodgeit.controllers.' + cname, None, None, [''])
controller = module.controller(req) controller = module.controller()
return getattr(controller, hname) return getattr(controller, hname)

89
lodgeit/controllers/pastes.py
View File

@ -8,12 +8,10 @@
:copyright: 2007 by Armin Ronacher. :copyright: 2007 by Armin Ronacher.
:license: BSD :license: BSD
""" """
import sqlalchemy as meta from lodgeit.utils import redirect, PageNotFound, Response, \
ctx, render_template
from lodgeit.application import render_template, redirect, PageNotFound, \
Response
from lodgeit.controllers import BaseController from lodgeit.controllers import BaseController
from lodgeit.database import Paste from lodgeit.database import db, Paste
from lodgeit.lib.highlighting import LANGUAGES, STYLES, get_style from lodgeit.lib.highlighting import LANGUAGES, STYLES, get_style
from lodgeit.lib.pagination import generate_pagination from lodgeit.lib.pagination import generate_pagination
@ -32,34 +30,35 @@ class PasteController(BaseController):
""" """
code = error = '' code = error = ''
language = 'text' language = 'text'
pastes = self.dbsession.query(Paste) pastes = db.session.query(Paste)
if self.request.method == 'POST': if ctx.request.method == 'POST':
code = self.request.form.get('code') code = ctx.request.form.get('code')
language = self.request.form.get('language') language = ctx.request.form.get('language')
try: try:
parent = pastes.selectfirst(Paste.c.paste_id == parent = pastes.filter(Paste.paste_id ==
int(self.request.form.get('parent'))) int(ctx.request.form.get('parent'))).first()
except (KeyError, ValueError, TypeError): except (KeyError, ValueError, TypeError):
parent = None parent = None
spam = self.request.form.get('webpage')# or \ spam = ctx.request.form.get('webpage')
#self.app.antispam.is_spam(code, language) #TODO: use AntiSpam again
# or \ #self.app.antispam.is_spam(code, language)
if spam: if spam:
error = 'contains spam' error = 'contains spam'
if code and language and not error: if code and language and not error:
paste = Paste(code, language, parent, self.request.user_hash) paste = Paste(code, language, parent, ctx.request.user_hash)
self.dbsession.save(paste) db.session.save(paste)
self.dbsession.flush() db.session.flush()
return redirect(paste.url) return redirect(paste.url)
else: else:
parent = self.request.args.get('reply_to') parent = ctx.request.args.get('reply_to')
if parent is not None and parent.isdigit(): if parent is not None and parent.isdigit():
parent = pastes.selectfirst(Paste.c.paste_id == parent) parent = pastes.filter(Paste.paste_id == parent).first()
code = parent.code code = parent.code
language = parent.language language = parent.language
return render_template(self.request, 'new_paste.html', return render_template('new_paste.html',
languages=LANGUAGES, languages=LANGUAGES,
parent=parent, parent=parent,
code=code, code=code,
@ -71,19 +70,19 @@ class PasteController(BaseController):
""" """
Show an existing paste. Show an existing paste.
""" """
linenos = self.request.args.get('linenos') != 'no' linenos = ctx.request.args.get('linenos') != 'no'
pastes = self.dbsession.query(Paste) pastes = db.session.query(Paste)
paste = pastes.selectfirst(Paste.c.paste_id == paste_id) paste = pastes.filter(Paste.c.paste_id == paste_id).first()
if paste is None: if paste is None:
raise PageNotFound() raise PageNotFound()
if raw: if raw:
return Response(paste.code, mimetype='text/plain; charset=utf-8') return Response(paste.code, mimetype='text/plain; charset=utf-8')
style, css = get_style(self.request) style, css = get_style(ctx.request)
paste.rehighlight(linenos) paste.rehighlight(linenos)
return render_template(self.request, 'show_paste.html', return render_template('show_paste.html',
paste=paste, paste=paste,
style=style, style=style,
css=css, css=css,
@ -101,10 +100,10 @@ class PasteController(BaseController):
""" """
Display the tree of some related pastes. Display the tree of some related pastes.
""" """
paste = Paste.resolve_root(self.dbsession, paste_id) paste = Paste.resolve_root(paste_id)
if paste is None: if paste is None:
raise PageNotFound() raise PageNotFound()
return render_template(self.request, 'paste_tree.html', return render_template('paste_tree.html',
paste=paste, paste=paste,
current=paste_id current=paste_id
) )
@ -113,28 +112,26 @@ class PasteController(BaseController):
""" """
Paginated list of pages. Paginated list of pages.
""" """
raise PageNotFound() raise PageNotFound() #XXX: activate again?
def link(page): def link(page):
if page == 1: if page == 1:
return '/all/' return '/all/'
return '/all/%d' % page return '/all/%d' % page
pastes = self.dbsession.query(Paste).select( pastes = db.session.query(Paste).order_by(
order_by=[meta.desc(Paste.c.pub_date)], Paste.c.pub_date.desc()
limit=10, ).limit(10).offset(10*(page-1))
offset=10 * (page - 1)
)
if not pastes and page != 1: if not pastes and page != 1:
raise PageNotFound() raise PageNotFound()
for paste in pastes: for paste in pastes:
paste.rehighlight() paste.rehighlight()
return render_template(self.request, 'show_all.html', return render_template('show_all.html',
pastes=pastes, pastes=pastes,
pagination=generate_pagination(page, 10, pagination=generate_pagination(page, 10,
Paste.count(self.request.engine), link), Paste.count(), link),
css=get_style(self.request)[1] css=get_style(ctx.request)[1]
) )
def compare_paste(self, new_id=None, old_id=None): def compare_paste(self, new_id=None, old_id=None):
@ -143,15 +140,15 @@ class PasteController(BaseController):
""" """
# redirect for the compare form box # redirect for the compare form box
if old_id is new_id is None: if old_id is new_id is None:
old_id = self.request.form.get('old', '-1').lstrip('#') old_id = ctx.request.form.get('old', '-1').lstrip('#')
new_id = self.request.form.get('new', '-1').lstrip('#') new_id = ctx.request.form.get('new', '-1').lstrip('#')
return redirect('/compare/%s/%s' % (old_id, new_id)) return redirect('/compare/%s/%s' % (old_id, new_id))
pastes = self.dbsession.query(Paste) pastes = db.session.query(Paste)
old = pastes.selectfirst(Paste.c.paste_id == old_id) old = pastes.filter(Paste.c.paste_id == old_id).first()
new = pastes.selectfirst(Paste.c.paste_id == new_id) new = pastes.filter(Paste.c.paste_id == new_id).first()
if old is None or new is None: if old is None or new is None:
raise PageNotFound() raise PageNotFound()
return render_template(self.request, 'compare_paste.html', return render_template('compare_paste.html',
old=old, old=old,
new=new, new=new,
diff=old.compare_to(new, template=True) diff=old.compare_to(new, template=True)
@ -161,9 +158,9 @@ class PasteController(BaseController):
""" """
Render an udiff for the two pastes. Render an udiff for the two pastes.
""" """
pastes = self.dbsession.query(Paste) pastes = db.session.query(Paste)
old = pastes.selectfirst(Paste.c.paste_id == old_id) old = pastes.filter(Paste.c.paste_id == old_id).first()
new = pastes.selectfirst(Paste.c.paste_id == new_id) new = pastes.filter(Paste.c.paste_id == new_id).first()
if old is None or new is None: if old is None or new is None:
raise PageNotFound() raise PageNotFound()
return Response(old.compare_to(new), mimetype='text/plain') return Response(old.compare_to(new), mimetype='text/plain')
@ -173,8 +170,8 @@ class PasteController(BaseController):
Minimal view that updates the style session cookie. Redirects Minimal view that updates the style session cookie. Redirects
back to the page the user is coming from. back to the page the user is coming from.
""" """
style_name = self.request.form.get('style') style_name = ctx.request.form.get('style')
resp = redirect(self.request.environ.get('HTTP_REFERER') or '/') resp = redirect(ctx.request.environ.get('HTTP_REFERER') or '/')
if style_name in STYLES: if style_name in STYLES:
resp.set_cookie('style', style_name) resp.set_cookie('style', style_name)
return resp return resp

20
lodgeit/controllers/static.py
View File

@ -8,7 +8,7 @@
:copyright: 2007 by Armin Ronacher. :copyright: 2007 by Armin Ronacher.
:license: BSD :license: BSD
""" """
from lodgeit.application import render_template, PageNotFound from lodgeit.utils import ctx, PageNotFound, render_template
from lodgeit.controllers import BaseController from lodgeit.controllers import BaseController
from lodgeit.lib.xmlrpc import xmlrpc from lodgeit.lib.xmlrpc import xmlrpc
@ -26,10 +26,10 @@ known_help_pages = set(x[0] for x in HELP_PAGES)
class StaticController(BaseController): class StaticController(BaseController):
def not_found(self): def not_found(self):
return render_template(self.request, 'not_found.html') return render_template('not_found.html')
def about(self): def about(self):
return render_template(self.request, 'about.html') return render_template('about.html')
def help(self, topic=None): def help(self, topic=None):
if topic is None: if topic is None:
@ -38,12 +38,14 @@ class StaticController(BaseController):
tmpl_name = 'help/%s.html' % topic tmpl_name = 'help/%s.html' % topic
else: else:
raise PageNotFound() raise PageNotFound()
return render_template(self.request, tmpl_name, return render_template(
help_topics=HELP_PAGES, tmpl_name,
current_topic=topic, help_topics=HELP_PAGES,
xmlrpc_url='http://%s/xmlrpc/' % current_topic=topic,
self.request.environ['SERVER_NAME'], xmlrpc_url='http://%s/xmlrpc/' %
xmlrpc_methods=xmlrpc.get_public_methods()) ctx.request.environ['SERVER_NAME'],
xmlrpc_methods=xmlrpc.get_public_methods()
)
controller = StaticController controller = StaticController

87
lodgeit/controllers/xmlrpc.py
View File

@ -8,11 +8,9 @@
:copyright: 2007 by Armin Ronacher, Georg Brandl. :copyright: 2007 by Armin Ronacher, Georg Brandl.
:license: BSD :license: BSD
""" """
import sqlalchemy as meta from lodgeit.utils import ctx, render_template
from lodgeit.application import render_template
from lodgeit.controllers import BaseController from lodgeit.controllers import BaseController
from lodgeit.database import Paste from lodgeit.database import db, Paste
from lodgeit.lib.xmlrpc import xmlrpc, exported from lodgeit.lib.xmlrpc import xmlrpc, exported
from lodgeit.lib.highlighting import STYLES, LANGUAGES, get_style, \ from lodgeit.lib.highlighting import STYLES, LANGUAGES, get_style, \
get_language_for get_language_for
@ -21,13 +19,13 @@ from lodgeit.lib.highlighting import STYLES, LANGUAGES, get_style, \
class XmlRpcController(BaseController): class XmlRpcController(BaseController):
def handle_request(self): def handle_request(self):
if self.request.method == 'POST': if ctx.request.method == 'POST':
return xmlrpc.handle_request(self.request) return xmlrpc.handle_request()
return render_template(self.request, 'xmlrpc.html') return render_template('xmlrpc.html')
@exported('pastes.newPaste') @exported('pastes.newPaste')
def pastes_new_paste(request, language, code, parent_id=None, def pastes_new_paste(language, code, parent_id=None,
filename='', mimetype=''): filename='', mimetype=''):
""" """
Create a new paste. Return the new ID. Create a new paste. Return the new ID.
@ -38,13 +36,13 @@ def pastes_new_paste(request, language, code, parent_id=None,
if not language: if not language:
language = get_language_for(filename or '', mimetype or '') language = get_language_for(filename or '', mimetype or '')
paste = Paste(code, language, parent_id) paste = Paste(code, language, parent_id)
request.dbsession.save(paste) db.session.save(paste)
request.dbsession.flush() db.session.flush()
return paste.paste_id return paste.paste_id
@exported('pastes.getPaste') @exported('pastes.getPaste')
def pastes_get_paste(request, paste_id): def pastes_get_paste(paste_id):
""" """
Get all known information about a paste by a given paste id. Get all known information about a paste by a given paste id.
@ -52,52 +50,51 @@ def pastes_get_paste(request, paste_id):
`paste_id`, `code`, `parsed_code`, `pub_date`, `language`, `paste_id`, `code`, `parsed_code`, `pub_date`, `language`,
`parent_id`, `url`. `parent_id`, `url`.
""" """
paste = request.dbsession.query(Paste).selectfirst(Paste.c.paste_id == paste = db.session.query(Paste).filter(Paste.c.paste_id ==
paste_id) paste_id).first()
if paste is None: if paste is None:
return False return False
return paste.to_xmlrpc_dict() return paste.to_xmlrpc_dict()
@exported('pastes.getDiff') @exported('pastes.getDiff')
def pastes_get_diff(request, old_id, new_id): def pastes_get_diff(old_id, new_id):
""" """
Compare the two pastes and return an unified diff. Compare the two pastes and return an unified diff.
""" """
paste = request.dbsession.query(Paste) pastes = db.session.query(Paste)
old = pastes.selectfirst(Paste.c.paste_id == old_id) old = pastes.filter(Paste.c.paste_id == old_id).first()
new = pastes.selectfirst(Paste.c.paste_id == new_id) new = pastes.filter(Paste.c.paste_id == new_id).first()
if old is None or new is None: if old is None or new is None:
return False return False
return old.compare_to(new) return old.compare_to(new)
@exported('pastes.getRecent') @exported('pastes.getRecent')
def pastes_get_recent(request, amount=5): def pastes_get_recent(amount=5):
""" """
Return information dict (see `getPaste`) about the last `amount` pastes. Return information dict (see `getPaste`) about the last `amount` pastes.
""" """
amount = min(amount, 20) amount = min(amount, 20)
return [x.to_xmlrpc_dict() for x in return [x.to_xmlrpc_dict() for x in
request.dbsession.query(Paste).select( db.session.query(Paste).order_by(
order_by=[meta.desc(Paste.c.pub_date)], Paste.pub_date.desc()
limit=amount ).limit(amount)]
)]
@exported('pastes.getLast') @exported('pastes.getLast')
def pastes_get_last(request): def pastes_get_last():
""" """
Get information dict (see `getPaste`) for the most recent paste. Get information dict (see `getPaste`) for the most recent paste.
""" """
rv = pastes_get_recent(request, 1) rv = pastes_get_recent(1)
if rv: if rv:
return rv[0] return rv[0]
return {} return {}
@exported('pastes.getLanguages') @exported('pastes.getLanguages')
def pastes_get_languages(request): def pastes_get_languages():
""" """
Get a list of supported languages. Get a list of supported languages.
""" """
@ -105,7 +102,7 @@ def pastes_get_languages(request):
@exported('styles.getStyles') @exported('styles.getStyles')
def styles_get_styles(request): def styles_get_styles():
""" """
Get a list of supported styles. Get a list of supported styles.
""" """
@ -113,7 +110,7 @@ def styles_get_styles(request):
@exported('styles.getStylesheet') @exported('styles.getStylesheet')
def styles_get_stylesheet(request, name): def styles_get_stylesheet(name):
""" """
Return the stylesheet for a given style. Return the stylesheet for a given style.
""" """
@ -121,48 +118,48 @@ def styles_get_stylesheet(request, name):
@exported('antispam.addRule', hidden=True) @exported('antispam.addRule', hidden=True)
def antispam_add_rule(request, rule): def antispam_add_rule(rule):
request.app.antispam.add_rule(rule) ctx.application.antispam.add_rule(rule)
@exported('antispam.removeRule', hidden=True) @exported('antispam.removeRule', hidden=True)
def antispam_remove_rule(request, rule): def antispam_remove_rule(rule):
request.app.antispam.remove_rule(rule) ctx.application.antispam.remove_rule(rule)
@exported('antispam.getRules', hidden=True) @exported('antispam.getRules', hidden=True)
def antispam_get_rules(request): def antispam_get_rules():
return sorted(request.app.antispam.get_rules()) return sorted(ctx.application.antispam.get_rules())
@exported('antispam.hasRule', hidden=True) @exported('antispam.hasRule', hidden=True)
def antispam_has_rule(request, rule): def antispam_has_rule(rule):
return request.app.antispam.rule_exists(rule) return ctx.application.antispam.rule_exists(rule)
@exported('antispam.addSyncSource', hidden=True) @exported('antispam.addSyncSource', hidden=True)
def antispam_add_sync_source(request, url): def antispam_add_sync_source(url):
request.app.antispam.add_sync_source(url) ctx.application.antispam.add_sync_source(url)
@exported('antispam.removeSyncSource', hidden=True) @exported('antispam.removeSyncSource', hidden=True)
def antispam_remove_sync_source(request, url): def antispam_remove_sync_source(url):
request.app.antispam.remove_sync_source(url) ctx.application.antispam.remove_sync_source(url)
@exported('antispam.getSyncSources', hidden=True) @exported('antispam.getSyncSources', hidden=True)
def antispam_get_sync_sources(request): def antispam_get_sync_sources():
return sorted(request.app.antispam.get_sync_sources()) return sorted(ctx.application.antispam.get_sync_sources())
@exported('antispam.hasSyncSource', hidden=True) @exported('antispam.hasSyncSource', hidden=True)
def antispam_has_sync_source(request, url): def antispam_has_sync_source(url):
return url in request.app.antispam.get_sync_sources() return url in ctx.application.antispam.get_sync_sources()
@exported('antispam.triggerSync', hidden=True) @exported('antispam.triggerSync', hidden=True)
def antispam_trigger_sync(request): def antispam_trigger_sync():
request.app.antispam.sync_sources() ctx.application.antispam.sync_sources()
controller = XmlRpcController controller = XmlRpcController

112
lodgeit/database.py
View File

@ -10,47 +10,65 @@
""" """
import time import time
import difflib import difflib
import sqlalchemy as meta from types import ModuleType
import sqlalchemy
from sqlalchemy import orm
from sqlalchemy.orm.scoping import ScopedSession
from cgi import escape from cgi import escape
from random import random
from hashlib import sha1
from datetime import datetime from datetime import datetime
from lodgeit.utils import _local_manager, ctx
from lodgeit.lib.highlighting import highlight, LANGUAGES from lodgeit.lib.highlighting import highlight, LANGUAGES
metadata = meta.MetaData()
def session_factory():
return orm.create_session(bind=ctx.application.engine)
session = ScopedSession(session_factory, scopefunc=_local_manager.get_ident)
#: create a fake module for easy access to database session methods
db = ModuleType('db')
key = value = mod = None
for mod in sqlalchemy, orm:
for key, value in mod.__dict__.iteritems():
if key in mod.__all__:
setattr(db, key, value)
del key, mod, value
db.__doc__ = __doc__
for name in 'delete', 'save', 'flush', 'execute', 'begin', 'query', \
'commit', 'rollback', 'clear', 'refresh', 'expire':
setattr(db, name, getattr(session, name))
db.session = session
pastes = meta.Table('pastes', metadata, metadata = db.MetaData()
meta.Column('paste_id', meta.Integer, primary_key=True),
meta.Column('code', meta.Unicode), pastes = db.Table('pastes', metadata,
meta.Column('parsed_code', meta.Unicode), db.Column('paste_id', db.Integer, primary_key=True),
meta.Column('parent_id', meta.Integer, meta.ForeignKey('pastes.paste_id'), db.Column('code', db.Unicode),
db.Column('parsed_code', db.Unicode),
db.Column('parent_id', db.Integer, db.ForeignKey('pastes.paste_id'),
nullable=True), nullable=True),
meta.Column('pub_date', meta.DateTime), db.Column('pub_date', db.DateTime),
meta.Column('language', meta.Unicode(30)), db.Column('language', db.Unicode(30)),
meta.Column('user_hash', meta.Unicode(40), nullable=True), db.Column('user_hash', db.Unicode(40), nullable=True),
meta.Column('handled', meta.Boolean, nullable=False) db.Column('handled', db.Boolean, nullable=False)
) )
spam_rules = meta.Table('spam_rules', metadata, spam_rules = db.Table('spam_rules', metadata,
meta.Column('rule_id', meta.Integer, primary_key=True), db.Column('rule_id', db.Integer, primary_key=True),
meta.Column('rule', meta.Unicode) db.Column('rule', db.Unicode)
) )
spamsync_sources = meta.Table('spamsync_sources', metadata, spamsync_sources = db.Table('spamsync_sources', metadata,
meta.Column('source_id', meta.Integer, primary_key=True), db.Column('source_id', db.Integer, primary_key=True),
meta.Column('url', meta.Unicode), db.Column('url', db.Unicode),
meta.Column('last_update', meta.DateTime) db.Column('last_update', db.DateTime)
) )
def generate_user_hash():
return sha1('%s|%s' % (random(), time.time())).hexdigest()
class Paste(object): class Paste(object):
def __init__(self, code, language, parent=None, user_hash=None): def __init__(self, code, language, parent=None, user_hash=None):
@ -113,36 +131,36 @@ class Paste(object):
return '<pre class="syntax">%s</pre>' % code return '<pre class="syntax">%s</pre>' % code
@staticmethod @staticmethod
def fetch_replies(req): def fetch_replies():
""" """
Get the new replies for the owern of a request and flag them Get the new replies for the ower of a request and flag them
as handled. as handled.
""" """
s = meta.select([pastes.c.paste_id], s = db.select([pastes.c.paste_id],
pastes.c.user_hash == req.user_hash Paste.user_hash == ctx.request.user_hash
)
paste_list = req.dbsession.query(Paste).select(
(Paste.c.parent_id.in_(s)) &
(Paste.c.handled == False) &
(Paste.c.user_hash != req.user_hash),
order_by=[meta.desc(Paste.c.pub_date)]
) )
paste_list = db.session.query(Paste).filter(db.and_(
Paste.parent_id.in_(s),
Paste.handled == False,
Paste.user_hash != ctx.request.user_hash,
)).order_by(pastes.c.paste_id.desc()).all()
to_mark = [p.paste_id for p in paste_list] to_mark = [p.paste_id for p in paste_list]
req.engine.execute(pastes.update(pastes.c.paste_id.in_(*to_mark)), db.execute(pastes.update(pastes.c.paste_id.in_(to_mark),
handled=True values={'handled': True}))
)
return paste_list return paste_list
@staticmethod @staticmethod
def count(con): def count():
s = meta.select([meta.func.count(pastes.c.paste_id)]) s = db.select([db.func.count(pastes.c.paste_id)])
return con.execute(s).fetchone()[0] return db.execute(s).fetchone()[0]
@staticmethod @staticmethod
def resolve_root(sess, paste_id): def resolve_root(paste_id):
q = sess.query(Paste) pastes = db.query(Paste)
while True: while True:
paste = q.selectfirst(Paste.c.paste_id == paste_id) paste = pastes.filter(Paste.c.paste_id == paste_id).first()
if paste is None: if paste is None:
return return
if paste.parent_id is None: if paste.parent_id is None:
@ -150,10 +168,10 @@ class Paste(object):
paste_id = paste.parent_id paste_id = paste.parent_id
meta.mapper(Paste, pastes, properties={ db.mapper(Paste, pastes, properties={
'children': meta.relation(Paste, 'children': db.relation(Paste,
primaryjoin=pastes.c.parent_id==pastes.c.paste_id, primaryjoin=pastes.c.parent_id==pastes.c.paste_id,
cascade='all', cascade='all',
backref=meta.backref('parent', remote_side=[pastes.c.paste_id]) backref=db.backref('parent', remote_side=[pastes.c.paste_id])
) )
}) })

33
lodgeit/lib/antispam.py
View File

@ -12,7 +12,7 @@ import re
import urllib import urllib
import time import time
from datetime import datetime, timedelta from datetime import datetime, timedelta
from lodgeit.database import spam_rules, spamsync_sources from lodgeit.database import db, spam_rules, spamsync_sources
UPDATE_INTERVAL = 60 * 60 * 24 * 7 UPDATE_INTERVAL = 60 * 60 * 24 * 7
@ -24,23 +24,22 @@ class AntiSpam(object):
updated from the moin moin server) and checks strings against it. updated from the moin moin server) and checks strings against it.
""" """
def __init__(self, app): def __init__(self):
self.engine = app.engine
self._rules = {} self._rules = {}
self.sync_with_db() self.sync_with_db()
def add_sync_source(self, url): def add_sync_source(self, url):
"""Add a new sync source.""" """Add a new sync source."""
self.engine.execute(spamsync_sources.insert(), url=url) db.session.execute(spamsync_sources.insert(), url=url)
def remove_sync_source(self, url): def remove_sync_source(self, url):
"""Remove a sync soruce.""" """Remove a sync soruce."""
self.engine.execute(spamsync_sources.delete( db.session.execute(spamsync_sources.delete(
spamsync_sources.c.url == url)) spamsync_sources.c.url == url))
def get_sync_sources(self): def get_sync_sources(self):
"""Get a list of all spamsync sources.""" """Get a list of all spamsync sources."""
return set(x.url for x in self.engine.execute( return set(x.url for x in db.session.execute(
spamsync_sources.select())) spamsync_sources.select()))
def add_rule(self, rule, noinsert=False): def add_rule(self, rule, noinsert=False):
@ -52,12 +51,12 @@ class AntiSpam(object):
return return
self._rules[rule] = regex self._rules[rule] = regex
if not noinsert: if not noinsert:
self.engine.execute(spam_rules.insert(), rule=rule) db.session.execute(spam_rules.insert(), rule=rule)
def remove_rule(self, rule): def remove_rule(self, rule):
"""Remove a rule from the database.""" """Remove a rule from the database."""
self._rules.pop(rule, None) self._rules.pop(rule, None)
self.engine.execute(spam_rules.delete(spam_rules.c.rule == rule)) db.session.execute(spam_rules.delete(spam_rules.c.rule == rule))
def get_rules(self): def get_rules(self):
"""Get a list of all spam rules.""" """Get a list of all spam rules."""
@ -71,7 +70,7 @@ class AntiSpam(object):
"""Sync with the database.""" """Sync with the database."""
# compile rules from the database and save them on the instance # compile rules from the database and save them on the instance
processed = set() processed = set()
for row in self.engine.execute(spam_rules.select()): for row in db.session.execute(spam_rules.select()):
if row.rule in processed: if row.rule in processed:
continue continue
processed.add(row.rule) processed.add(row.rule)
@ -83,14 +82,14 @@ class AntiSpam(object):
for rule in set(self._rules) - processed: for rule in set(self._rules) - processed:
del self._rules[rule] del self._rules[rule]
to_delete.append(rule) to_delete.append(rule)
self.engine.execute(spam_rules.delete( db.session.execute(spam_rules.delete(
spam_rules.c.rule.in_(*to_delete))) spam_rules.c.rule.in_(to_delete)))
# otherwise add the rules to the database # otherwise add the rules to the database
else: else:
for rule in set(self._rules) - processed: for rule in set(self._rules) - processed:
self.engine.execute(spam_rules.insert(), db.session.execute(spam_rules.insert(),
rule=rule) rule=rule)
def sync_sources(self): def sync_sources(self):
"""Trigger the syncing.""" """Trigger the syncing."""
@ -103,7 +102,7 @@ class AntiSpam(object):
update_after = datetime.utcnow() - timedelta(seconds=UPDATE_INTERVAL) update_after = datetime.utcnow() - timedelta(seconds=UPDATE_INTERVAL)
q = (spamsync_sources.c.last_update == None) | \ q = (spamsync_sources.c.last_update == None) | \
(spamsync_sources.c.last_update < update_after) (spamsync_sources.c.last_update < update_after)
sources = list(self.engine.execute(spamsync_sources.select(q))) sources = list(db.session.execute(spamsync_sources.select(q)))
if sources: if sources:
for source in sources: for source in sources:
self.sync_source(source.url) self.sync_source(source.url)
@ -113,8 +112,8 @@ class AntiSpam(object):
"""Sync one source.""" """Sync one source."""
self.load_rules(url) self.load_rules(url)
q = spamsync_sources.c.url == url q = spamsync_sources.c.url == url
self.engine.execute(spamsync_sources.update(q), db.session.execute(spamsync_sources.update(q),
last_update=datetime.utcnow()) last_update=datetime.utcnow())
def load_rules(self, url): def load_rules(self, url):
"""Load some rules from an URL.""" """Load some rules from an URL."""

8
lodgeit/lib/xmlrpc.py
View File

@ -11,8 +11,8 @@
import re import re
import inspect import inspect
from SimpleXMLRPCServer import SimpleXMLRPCDispatcher from SimpleXMLRPCServer import SimpleXMLRPCDispatcher
from lodgeit.utils import ctx, Response
from lodgeit.application import Response
_strip_re = re.compile(r'[\x00-\x08\x0B-\x1F]') _strip_re = re.compile(r'[\x00-\x08\x0B-\x1F]')
@ -26,10 +26,10 @@ class XMLRPCRequestHandler(SimpleXMLRPCDispatcher):
def list_methods(self, request): def list_methods(self, request):
return [x['name'] for x in self.get_public_methods()] return [x['name'] for x in self.get_public_methods()]
def handle_request(self, request): def handle_request(self):
def dispatch(method_name, params): def dispatch(method_name, params):
return self.funcs[method_name](request, *params) return self.funcs[method_name](*params)
response = self._marshaled_dispatch(request.data, dispatch) response = self._marshaled_dispatch(ctx.request.data, dispatch)
return Response(response, mimetype='text/xml') return Response(response, mimetype='text/xml')
def get_public_methods(self): def get_public_methods(self):

44
lodgeit/urls.py
View File

@ -10,30 +10,28 @@
""" """
from werkzeug.routing import Map, Rule from werkzeug.routing import Map, Rule
urlmap = Map( urlmap = Map([
[ # paste interface
# paste interface Rule('/', endpoint='pastes/new_paste'),
Rule('/', endpoint='pastes/new_paste'), Rule('/show/<int:paste_id>/', endpoint='pastes/show_paste'),
Rule('/show/<int:paste_id>/', endpoint='pastes/show_paste'), Rule('/raw/<int:paste_id>/', endpoint='pastes/raw_paste'),
Rule('/raw/<int:paste_id>/', endpoint='pastes/raw_paste'), Rule('/compare/', endpoint='pastes/compare_paste'),
Rule('/compare/', endpoint='pastes/compare_paste'), Rule('/compare/<int:new_id>/<int:old_id>/', endpoint='pastes/compare_paste'),
Rule('/compare/<int:new_id>/<int:old_id>/', endpoint='pastes/compare_paste'), Rule('/unidiff/<int:new_id>/<int:old_id>/', endpoint='pastes/unidiff_paste'),
Rule('/unidiff/<int:new_id>/<int:old_id>/', endpoint='pastes/unidiff_paste'), Rule('/tree/<int:paste_id>/', endpoint='pastes/show_tree'),
Rule('/tree/<int:paste_id>/', endpoint='pastes/show_tree'),
# paste list # paste list
Rule('/all/', endpoint='pastes/show_all'), Rule('/all/', endpoint='pastes/show_all'),
Rule('/all/<int:page>/', endpoint='pastes/show_all'), Rule('/all/<int:page>/', endpoint='pastes/show_all'),
# xmlrpc # xmlrpc
Rule('/xmlrpc/', endpoint='xmlrpc/handle_request'), Rule('/xmlrpc/', endpoint='xmlrpc/handle_request'),
# static pages # static pages
Rule('/about/', endpoint='static/about'), Rule('/about/', endpoint='static/about'),
Rule('/help/', endpoint='static/help'), Rule('/help/', endpoint='static/help'),
Rule('/help/<topic>/', endpoint='static/help'), Rule('/help/<topic>/', endpoint='static/help'),
# colorscheme # colorscheme
Rule('/colorscheme/', endpoint='pastes/set_colorscheme'), Rule('/colorscheme/', endpoint='pastes/set_colorscheme'),
], ])
)

117
lodgeit/utils.py Normal file
View File

@ -0,0 +1,117 @@
# -*- coding: utf-8 -*-
"""
lodgeit.utils
~~~~~~~~~~~~~
Serveral utilities used by LodgeIt.
:copyright: 2007 by Christopher Grebs.
:license: BSD
"""
import time
from hashlib import sha1
from random import random
from types import ModuleType
from werkzeug import Local, LocalManager, LocalProxy, BaseRequest, \
BaseResponse
from werkzeug.routing import NotFound, RequestRedirect
from jinja import Environment, PackageLoader
#: context locals
_local = Local()
_local_manager = LocalManager([_local])
#: fake module type for easy access to context locals
ctx = ModuleType('ctx')
ctx.__doc__ = 'Module that holds all context locals'
#: some variables for direct access to the context locals
ctx.application = LocalProxy(_local, 'application')
ctx.request = LocalProxy(_local, 'request')
#: jinja environment for all the templates
jinja_environment = Environment(loader=PackageLoader('lodgeit', 'views',
use_memcache=False,
auto_reload=False
))
def datetimeformat():
"""
Helper filter for the template
"""
def wrapped(env, ctx, value):
return value.strftime('%Y-%m-%d %H:%M')
return wrapped
jinja_environment.filters['datetimeformat'] = datetimeformat
def generate_user_hash():
return sha1('%s|%s' % (random(), time.time())).hexdigest()
class Request(BaseRequest):
"""
Subclass of the `BaseRequest` object. automatically creates a new
`user_hash` and sets `first_visit` to `True` if it's a new user.
It also stores the engine and dbsession on it.
"""
charset = 'utf-8'
def __init__(self, environ):
self.app = ctx.application
super(Request, self).__init__(environ)
# check the user hash. an empty cookie is considered
# begin a new session.
self.user_hash = ''
self.first_visit = False
if 'user_hash' in self.cookies:
self.user_hash = self.cookies['user_hash']
if not self.user_hash:
self.user_hash = generate_user_hash()
self.first_visit = True
def bind_to_context(self):
ctx.request = self
class Response(BaseResponse):
"""
Subclass the response object for later extension.
"""
charset = 'utf-8'
class PageNotFound(NotFound):
"""
Internal exception used to tell the application to show the
error page.
"""
def render_template(template_name, **tcontext):
"""
Render a template to a response. This automatically fetches
the list of new replies for the layout template. It also
adds the current request to the context. This is used for the
welcome message.
"""
from lodgeit.database import Paste
if ctx.request.method == 'GET':
tcontext['new_replies'] = Paste.fetch_replies()
tcontext['request'] = ctx.request
t = jinja_environment.get_template(template_name)
return Response(t.render(tcontext), mimetype='text/html; charset=utf-8')
def redirect(url, code=302):
"""
Redirect to somewhere. Returns a nice response object.
"""
return Response('Page Moved to %s' % url,
headers=[('Location', url),
('Content-Type', 'text/plain')],
status=code)

35
runlodgeit.py
View File

@ -1,6 +1,37 @@
from lodgeit.application import make_app from lodgeit.application import make_app
from lodgeit.utils import ctx
from lodgeit.database import db
from werkzeug import script
from werkzeug.debug import DebuggedApplication from werkzeug.debug import DebuggedApplication
from werkzeug.serving import run_simple from werkzeug.serving import run_simple
from werkzeug.utils import create_environ, run_wsgi_app
app = DebuggedApplication(make_app('sqlite:////tmp/lodgeit.db')) dburi = 'sqlite:////tmp/lodgeit.db'
run_simple('localhost', 7000, app, True)
def run_app(app, path='/'):
env = create_environ(path)
return run_wsgi_app(app, env)
action_runserver = script.make_runserver(
lambda: make_app(dburi),
use_reloader=True)
action_rundserver = script.make_runserver(
lambda: DebuggedApplication(make_app(dburi, True), evalex=True),
use_reloader=True)
action_shell = script.make_shell(
lambda: {
'app': make_app(dburi, False, True),
'ctx': ctx,
'db': db,
'run_app': run_app
},
('\nWelcome to the interactive shell environment of LodgeIt!\n'
'\n'
'You can use the following predefined objects: app, ctx, db.\n'
'To run the application (creates a request) use *run_app*.')
)
if __name__ == '__main__':
script.run()