Skip to main content

API Decorators

Project description

API Decorators

define REST schemas for aiohttp and mongodb.

pip install apidecorators

This API is valid for schemas in this way:

schema = { * primitives * objects * arrays of primitives * arrays of objects }

where objects are objects of primitives and this kind of objects.

Let me explain by examples:

from apidecorators.api import jwt_auth, get, insert, has_role, update, push, pull, \
                validate, get_many, delete, read_access, write_access, collection, aggregate, \
                update_array, get_from_array, public
from apidecorators.fields import all_fields            
from cerberus import Validator

#given this schema

s_budget = {
    '_id': {'type': 'string'},
    'applicant': {'type': 'string'},
    'offerer': {'type': 'string'},
    'description': {'type': 'string', 'required': True},
    'amount': {'type': 'integer'},
    'favorite': {'type': 'boolean'},
    'comment': {'type': 'dict',
                'schema': {
                    "text": {"type": "string"},
                    "date": {"type": "float"}

s_demand = {
    '_id': {'type': 'string'},
    'applicant': {'type': 'string'},
    'description': {'type': 'string', 'required': True},
    'location': {'type': 'string'},
    'budgets': {
        'type': 'list',
        'schema': s_budget

v_demand = Validator(s_demand)
v_budget = Validator(s_budget)

# we define a POST this way

def set_routes_demand(routes):'/api/demand') # aiohttp routes
    @jwt_auth # must receive a valid JWT token
    @collection('demand') # which collection
    @write_access({'*': '*'}) # any user can write any field
    @validate(validator=v_demanda) # set the cerberus validator
    @insert # it will be an insert
    async def post_demand(document, request, token):
        document['applicant'] = token['user']
        return document  # the returned document will be written in the collection described above

# we GET a document this way:

    # the user stores in the field applicant can read all fields minus location
    # any other user can read only description and location
    @read_access({'applicant': all_fields(s_demand) - {'location'}, '*': {'description', 'location'}})
    @get # it will be a get
    async def get_demand(document, token):
        # the last chance to change the document that will be sent to the client    
        return document

# we PUT a document this way:

    @write_access({'applicant': {'description', 'location'}})
    @validate(update=True, validator=v_demanda) # see the attribute update=True
    @update # it will be an update
    async def put_demand(old_doc, document, request, token):      
        return document

# let see how to push to an array:

    @write_access({'*': {'description', 'amount'}})
    @push('budgets') # the name of the array
    async def push_budget(old_doc, document, request, token):
        document['offerer'] = token['user']
        document['applicant'] = old_doc['applicant']      
        return document

# update an element of an array

    # the user stores in offerer field of subdocument (sub_id) can update description and amount
    # the user stores in applicant field of subdocument (sub_id) can update favorite and comment
    # if you pass root a value different from '.', that will be the root where to check users of write_access
    @write_access({'offerer': {'description', 'amount'}, 'applicant': {'favorite', 'comment'}}, root='budgets')
    async def update_budgets(old_doc, document, token):      
        return document

# get many

    @read_access({'*': '*'})
    async def get_many_demands(col, query, token):  
        applicant = query["applicant"]
        return col.find({"applicant": applicant}).skip(0).limit(10)

# get from an array

    @read_access({'offerer': {'description', 'amount'}})
    async def get_presupuestos(document, token):   
        #the chance to remove empty objects in array   
        return document

# and you can do aggregates

    @public # it is not restricted by a JWT token, the user will be anonymous
    async def get_aggr_comments(col, query, token):  
        offerer = query["offerer"]
        pipeline = [
    {"$match": {"budgets.offerer": offerer}},
    {"$unwind": "$budgets"},
    {"$match": {"budgets.offerer": offerer}},
    {"$group": {"_id": "$budgets.offerer", "comments": {"$push": {
        "text": "$budgets.comment.text",
        "date": "$",
        "author": "$applicant"
        return col.aggregate(pipeline)

In read_access and write_access you can use dot notation and the $ for the array. Example:

@read_access({'applicant': all_fields(s_demand) | {'budgets.$.comment'}})
import asyncio
from demand import set_routes_demand
from aiohttp import web
from apidecorators.api import cors_factory

async def handle(loop):
    app = web.Application(loop=loop, middlewares=[cors_factory])
    routes = web.RouteTableDef()

    await loop.create_server(app.make_handler(), '', 8888)

def main():    
    loop = asyncio.get_event_loop()
    print("Server started at port 8888")

if __name__ == '__main__':


    - DB_URI=mongodb://<user>:<password>@url:port/data-base
    - DB=data-base
    - SECRET=secret

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for apidecorators, version 0.2.6
Filename, size File type Python version Upload date Hashes
Filename, size apidecorators-0.2.6-py3-none-any.whl (7.5 kB) File type Wheel Python version py3 Upload date Hashes View
Filename, size apidecorators-0.2.6.tar.gz (6.5 kB) File type Source Python version None Upload date Hashes View

Supported by

Pingdom Pingdom Monitoring Google Google Object Storage and Download Analytics Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page