Source code for yarhp.sqla

#!/usr/bin/python
# -*- coding: utf-8 -*-
from decimal import Decimal

from pyramid.httpexceptions import HTTPOk, HTTPNotFound, HTTPConflict, \
    HTTPCreated
from sqlalchemy.exc import IntegrityError

from yarhp import paginate, wrappers
from yarhp.view import BaseView
from yarhp.utils import get_composite_key


def get_model_schema(model):
    """Returns a validation schema for the ``model``
    the schema is a dict of dicts in the form of:
        
        .. code-block:: python

            {field_name1:{
                'type':<python type>,
                'required':<bool>,
                'length':<int>}
            field_name2:{
                ...
            }
            ...
        }
    """

    from sqlalchemy import types as sqla_types
    from datetime import date, datetime

    type_map = {
        sqla_types.Integer: int,
        sqla_types.Numeric: Decimal,
        sqla_types.BigInteger: int,
        sqla_types.Boolean: bool,
        sqla_types.Float: float,
        sqla_types.String: str,
        sqla_types.Text: str,
        sqla_types.CHAR: str,
        sqla_types.Enum: unicode,
        sqla_types.Unicode: unicode,
        sqla_types.Date: date,
        sqla_types.DateTime: datetime,
    }
    schema = {}

    for name in model.mapper.columns.keys():
        col = model.mapper.get_property(name).columns[0]
        schema[name] = dict(type=type_map.get(type(col.type)),
                            required=(not col.nullable and not col.server_default),
                            length=getattr(col.type, 'length', None))
    return schema


class SQLAView(BaseView):
    """Yarhp view used to serve SQLAlchemy objects.
    ``SQLAView implements the following logic:
        default behaviour for all actions
        validation agaisnt the sqla model's schema
    """

    _default_actions = frozenset('index show delete create update'.split())

    def validation_schema(self):
        "Returns the validation schema dict"

        schema = get_model_schema(self.__model_class__)
        # update the model's schema with the one supplied in view's __validation_schema__
        schema.update(self.__validation_schema__)
        self.__validation_schema__ = schema
        return self.__validation_schema__

    def default_index(self, **kw):
        """default for index action.
        Returns sqla query object which is passed to the pager to paginate the results.
        """

        return self.__model_class__.query.filter_by(**kw)

    def default_show(self, **kw):
        """default for show action.
        Raises HTTPNotFound if the row is not found.
        Returns the resouce as python dict.
        """

        key = get_composite_key(self.__model_class__, kw)
        self.active_obj = self.__model_class__.get(key)

        if not self.active_obj:
            raise HTTPNotFound()

        return self.active_obj.to_dict()

    def default_delete(self, **kw):
        """default for delete action.
        Raises HTTPNotFound if the row is not found.
        Returns HTTPOk on success.
        """

        key = get_composite_key(self.__model_class__, kw)
        self.active_obj = self.__model_class__.get(key)
        if not self.active_obj:
            raise HTTPNotFound()

        self.active_obj.delete()
        return HTTPOk()

    def default_create(self, **kw):
        """default for create action.
        Raises HTTPConflict if there is an integrity error in DB.
        Returns HTTPCreated on success.
        """

        params = self.request.params.copy()
        params.update(kw)
        self.active_obj = self.__model_class__(**params)

        def location(params):
            return self.request.route_url(self.resource.action_route_map['show'],
                    **params)

        try:
            self.active_obj.save()
            params['id'] = self.active_obj.id
            return HTTPCreated(location=location(params))
        except IntegrityError:
            raise HTTPConflict(location=location(params))

    def default_update(self, **kw):
        """default for update action.
        Raises HTTPNotFound if the resource being updated is not found.
        Returns HTTPOk on success.
        """

        key = get_composite_key(self.__model_class__, kw)
        self.active_obj = self.__model_class__.get(key)
        if not self.active_obj:
            raise HTTPNotFound()

        params = self.request.params.copy()
        params.update(kw)
        self.active_obj.update(**params)
        self.active_obj.save()

        return HTTPOk()