mediagoblin/db/sql/base.py

   1 # GNU MediaGoblin -- federated, autonomous media hosting
   2 # Copyright (C) 2011,2012 MediaGoblin contributors.  See AUTHORS.
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published by
   6 # the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17
  18 from sqlalchemy.ext.declarative import declarative_base
  19 from sqlalchemy.orm import scoped_session, sessionmaker, object_session
  20 from sqlalchemy.orm.query import Query
  21 from sqlalchemy.sql.expression import desc
  22 from mediagoblin.db.sql.fake import DESCENDING
  23
  24
  25 def _get_query_model(query):
  26     cols = query.column_descriptions
  27     assert len(cols) == 1, "These functions work only on simple queries"
  28     return cols[0]["type"]
  29
  30
  31 class GMGQuery(Query):
  32     def sort(self, key, direction):
  33         key_col = getattr(_get_query_model(self), key)
  34         if direction is DESCENDING:
  35             key_col = desc(key_col)
  36         return self.order_by(key_col)
  37
  38     def skip(self, amount):
  39         return self.offset(amount)
  40
  41
  42 Session = scoped_session(sessionmaker(query_cls=GMGQuery))
  43
  44
  45 def _fix_query_dict(query_dict):
  46     if '_id' in query_dict:
  47         query_dict['id'] = query_dict.pop('_id')
  48
  49
  50 class GMGTableBase(object):
  51     query = Session.query_property()
  52
  53     @classmethod
  54     def find(cls, query_dict={}):
  55         _fix_query_dict(query_dict)
  56         return cls.query.filter_by(**query_dict)
  57
  58     @classmethod
  59     def find_one(cls, query_dict={}):
  60         _fix_query_dict(query_dict)
  61         return cls.query.filter_by(**query_dict).first()
  62
  63     @classmethod
  64     def one(cls, query_dict):
  65         return cls.find(query_dict).one()
  66
  67     def get(self, key):
  68         return getattr(self, key)
  69
  70     def save(self, validate=True):
  71         assert validate
  72         sess = object_session(self)
  73         if sess is None:
  74             sess = Session()
  75         sess.add(self)
  76         sess.commit()
  77
  78
  79 Base = declarative_base(cls=GMGTableBase)
  80
  81
  82 class DictReadAttrProxy(object):
  83     """
  84     Maps read accesses to obj['key'] to obj.key
  85     and hides all the rest of the obj
  86     """
  87     def __init__(self, proxied_obj):
  88         self.proxied_obj = proxied_obj
  89
  90     def __getitem__(self, key):
  91         try:
  92             return getattr(self.proxied_obj, key)
  93         except AttributeError:
  94             raise KeyError("%r is not an attribute on %r"
  95                 % (key, self.proxied_obj))