Skip to content

Instantly share code, notes, and snippets.

@anandology
Last active August 29, 2015 14:07
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save anandology/afa2968be5b92f5637e7 to your computer and use it in GitHub Desktop.
Save anandology/afa2968be5b92f5637e7 to your computer and use it in GitHub Desktop.
SA like query chaining for web.py db
import web
class SelectQuery:
def __init__(self, tables):
if isinstance(tables, basestring):
tables = [tables]
self._tables = tables
self._columns = "*"
self._where = web.SQLQuery("")
self._order = []
self._limit = None
self._offset = None
def _copy(self):
q = SelectQuery(self._tables)
q._columns = self._columns
q._where = self._where
q._order = list(self._order) # make a copy
q._limit = self._limit
q._offset = self._offset
return q
def filter(self, condition, vars={}):
q = self._copy()
q._where = self._where + " AND " + web.reparam(condition, vars)
return q
def join(self, table, condition, vars={}):
q = self._copy()
q._tables.append(table)
return q.filter(condition, vars)
def order_by(self, column):
self._order.append(column)
def limit(self, n):
self._limit = n
def execute(self, db):
if self.order:
order = ",".join(self._order)
else:
order = None
return db.select(self._tables,
what=self._columns,
where=self._where,
order=self._order,
limit=self._limit,
offset=self._offset)
def getposts(basequery=None, pinned=False, showall=False, statuses=None):
if not statuses:
statuses = POSTSTATUS.LISTED
if basequery is None:
basequery = SelectQuery("jobpost")
query = basequery.filter("jobpost.status IN $statues", vars=locals())
if showall:
query = query.filter("jobpost.datetime > $d", vars=dict(d=datetime.utcnow() - agelimit))
else:
query = query.filter("(" +
"(jobpost.pinned = $t AND jobpost.datetime > $d1) OR" +
"(jobpost.pinned = $f AND jobpost.datetime > $d2)",
vars=dict(t=True,
f=False,
d1=datetime.utcnow() - agelimit,
d2=datetime.utcnow() - newlimit))
if g.board:
query = query.join("board_jobpost", "board_jobpost.jobpost=jobpost.id")
query = query.filter("board_jobpost.board_id = $g.board", vars=dict(g=g))
if pinned:
if g.board:
query = query.order_by("board_jobpost.pinned DESC")
else:
query = query.order_by("jobpost.pinned DESC")
return query.order_by("jobpost.datetime DESC")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment