With Pyramid

Since version 1.7 there are some Pyramid specific subclasses that help using the proxies within a Pyramid view as well as a expose decorator that simplify their implementation.

class metapensiero.sqlalchemy.proxy.pyramid.expose(proxable, metadata=None, adaptor=None, POST=True, **kw)

Decorator to simplify exposition of a SQLAlchemy Query.

Parameters:

• proxable – either a SQLAlchemy Query or a mapped class

• metadata – a dictionary with additional info about the fields

• adaptor – if given, it’s a function that will be called to adapt incoming data before actually writing it to the database.

• POST (either a boolean flag or a function, True by default) – whether to handle POST request: if True a standard function will be used, otherwise it must be a function accepting two positional arguments, respectively the SQLAlchemy session and the Pyramid request object, and a set of keyword arguments corresponding to the changed field

This is an helper class that aids the exposition of either a SQLAlchemy Query or directly a mapped class as a Pyramid view.

User of this class must inject a concrete implementation of the create_session() and save_changes() static methods. This is usually done once at application startup, for example:

from ..models import DBSession
from ..models.utils import save_changes

# Configure the `expose` decorator
expose.create_session = staticmethod(lambda req: DBSession())
expose.save_changes = staticmethod(save_changes)

Another class method that may eventually be replaced is extract_parameters(): the default implementation simply returns a copy of the request.params dictionary, but sometimes it is desiderable to pass additional parameters, for example when using bindparams:

def _extract_parameters(request):
    "Build a dictionary of arguments for the proxy from the current request"

    parameters = dict(request.params)
    # The following feeds eventual `bindparams`
    parameters['params'] = dict(request.session)
    return parameters

expose.extract_parameters = staticmethod(_extract_parameters)

The typical usage is:

@view_config(route_name='users', renderer='json')
@expose(User, metadata=dict(
    password=dict(hidden=True, password=True, width=40),
    is_anonymous=False,
    ))
def users(request, results):
    return results

The first argument may be either a mapped class or a query.

The decorated function is finally called with the current request and the result of the operation, and it can eventually adjust the results dictionary.

The decorated function may be a generator instead, which has the opportunity of freely manipulate either the arguments received from the request, or the final result, or both as follows:

@expose(User, metadata=dict(
    password=dict(hidden=True, password=True, width=40),
    is_anonymous=False,
    ))
def complex():
    # Receive request and params
    request, params = (yield)
    log.debug('REQUEST: %r', request)

    # Adjust parameters
    params['new'] = True

    if 'something' in params:
        # Inject other conditions
        something = params.pop('something')
        conditions = (User.c.foo == something,)
        result = yield params, conditions
    else:
        # Go on, and receive the final result
        result = yield params

    # Fix it up
    result['COMPLEX'] = 'MAYBE'

    yield result

As you can see, in this case the decorated function shall not declare any formal argument, because it receives its “feed” as the result of the yield expressions.

static create_session(request)

Create a new SQLAlchemy session, given the current request.

static extract_parameters(request)

Create a dictionary of parameters from the current request.

static save_changes(sa_session, modified, deleted)

Save insertions, changes and deletions to the database.

Parameters:

• sa_session – the SQLAlchemy session

• modified – a sequence of record changes, each represented by a tuple of two items, the PK name and a dictionary with the modified fields; if the value of the PK field is null or 0 then the record is considered new and will be inserted instead of updated

• deleted – a sequence of deletions, each represented by a tuple of two items, the PK name and the ID of the record to be removed

Return type:

a tuple of three lists, respectively inserted, modified and deleted record IDs, grouped in a dictionary keyed on PK name.

Basic setup

First of all, there are some setup steps to follow:

1. Include the package in the configuration file:

   [app:main]
   use = egg:ph.data
   
   ...
   
   pyramid.includes =
       metapensiero.sqlalchemy.proxy.pyramid
       pyramid_tm
   

   This is not strictly needed, but it will override the standard json renderer with one that uses python-rapidjson, to handle the datetime type.

2. Configure the expose decorator, for example adding something like the following snippet to the .../views.py source:

   from metapensiero.sqlalchemy.proxy.pyramid import expose
   from .models import DBSession
   
   # Configure the `expose` decorator
   expose.create_session = staticmethod(lambda req: DBSession())
   

Then you can add views to expose either an entity or a plain select:

@view_config(route_name='users', renderer='json')
@expose(User, metadata=dict(
    password=dict(hidden=True, password=True, width=40),
    active=dict(default=True),
    is_anonymous=False,
    ))
def users(request, results):
    return results

sessions_t = Session.__table__

@view_config(route_name='sessions', renderer='json')
@expose(select([sessions_t], sessions_t.c.iduser == bindparam('user_id')))
def sessions(request, results):
    return results

The decorated function may be a generator instead, which has the opportunity of freely manipulate either the arguments received from the request, or the final result, or both as follows:

@view_config(route_name='users', renderer='json')
@expose(User, metadata=dict(
    password=dict(hidden=True, password=True, width=40),
    active=dict(default=True),
    is_anonymous=False,
    ))
def complex():
    # Receive request and arguments
    request, args = (yield)

    # Adjust parameters
    args['new'] = True

    # Note that bindparams by default are extracted from the “params”
    # keyword argument
    bindparams = args.setdefault('params', {})
    bindparams['user_id'] = 2

    if 'something' in params:
        # Inject other conditions
        something = args.pop('something')
        conditions = (User.c.foo == something,)
        result = yield args, conditions
    else:
        # Go on, and receive the final result
        result = yield args

    # Fix it up
    result['COMPLEX'] = 'MAYBE'

    yield result

Request examples

Assuming the users view is added as /views/users, it could be called in the following ways:

GET /views/users

would return a JSON response containing all users, like:

{
  "count": 1234,
  "message": "Ok",
  "success": true,
  "root": [
    {
      "first_name": "Lele",
      "last_name": "Gaifax",
      ...
    },
    {
      "first_name": "Mario",
      "last_name": "Rossi",
      ...
    },
    ...
  ]
}

GET /views/users?limit=1&start=2

would return a JSON response containing just one user, the second:

{
  "count": 1234,
  "message": "Ok",
  "success": true,
  "root": [
    {
      "first_name": "Mario",
      "last_name": "Rossi",
      ...
    }
  ]
}

GET /views/users?filter_by_first_name=Lele

would return a JSON response containing the records satisfying the given condition:

{
  "count": 1,
  "message": "Ok",
  "success": true,
  "root": [
    {
      "first_name": "Lele",
      "last_name": "Gaifax",
      ...
    }
  ]
}

GET /views/users?filter_col=first_name&filter_value=Lele

same as above

GET /views/users?limit=1&only_cols=first_name,role_name

would return a JSON response containing only the requested fields of a single record:

{
  "count": 1234,
  "message": "Ok",
  "success": true,
  "root": [
    {
      "first_name": "Lele",
      "role_name": "administrator"
    }
  ]
}

GET /views/users?metadata=metadata&limit=0

would return a JSON response containing a description of the schema:

{
  "metadata": {
    "success_slot": "success",
    "primary_key": "iduser",
    "fields": [
      {
        "width": 60,
        "hint": "The unique ID of the user.",
        "align": "right",
        "nullable": false,
        "readonly": true,
        "type": "int",
        "hidden": true,
        "label": "User ID",
        "name": "iduser"
      },
      {
        "type": "boolean",
        "label": "Active",
        "hint": "Whether the user is currently active.",
        "name": "active",
        "default": true
        ...
      }
      ...
    ],
    "root_slot": "root",
    "count_slot": "count"
  },
  "message": "Ok",
  "success": true
}

Browse SoL sources for real usage examples.

Obtaining original instances

By default the decorator configures the proxy to return plain Python dictionaries, but sometime you may need to manipulate the resultset in way that it is easier done working with the actual instances.

In such cases you can explicitly pass asdict=False and then convert the list of instances to a JSON-serializable list of dictionaries by yourself, for example with something like the following, where the Image class is using sqlalchemy-media to store image details in the image field, backed by a JSONB column, of the class:

@view_config(route_name='images', renderer='json')
@expose(
    Image,
    asdict=False, fields=('id', 'position', 'width', 'height', 'original_filename'),
    metadata=dict(
        height=dict(label=_('Height'),
                    hint=_('The height of the image.')),
        original_filename=dict(label=_('Name'),
                               hint=_('Original name of the image.')),
        width=dict(label=_('Width'),
                   hint=_('The width of the image.'))))
def listing(request, results):
    root = results.get('root')
    if root:
        results['root'] = newroot = []
        for image in root:
            img = image.image
            d = {'id': image.id, 'position': image.position,
                 'width': img['width'], 'height': img['height'],
                 'original_filename': img['original_filename']}
            thumbnail = img.get_thumbnail(width=200, auto_generate=True)
            d['image_url'] = thumbnail.locate()
            newroot.append(d)
    return results