marshmallow_sample.py

# -*- coding:utf-8 -*-
import logging
logger = logging.getLogger(__name__)
from datetime import datetime
from pyramid.config import Configurator
from pyramid.view import view_config
from pyramid.httpexceptions import HTTPError
from pyramid.response import Response
import json
import sqlalchemy as sa
import sqlalchemy.orm as orm
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker
from marshmallow import Schema, fields, ValidationError
from zope.sqlalchemy import ZopeTransactionExtension

Session = scoped_session(sessionmaker(autoflush=False, extension=ZopeTransactionExtension()))
Base = declarative_base()


# model
class Author(Base):
    __tablename__ = "authors"
    query = Session.query_property()

    id = sa.Column(sa.Integer, primary_key=True)
    first = sa.Column(sa.String(80))
    last = sa.Column(sa.String(80))

    def __init__(self, first, last):
        self.first = first
        self.last = last


class Quote(Base):
    __tablename__ = "qoutes"
    query = Session.query_property()

    id = sa.Column(sa.Integer, primary_key=True)
    content = sa.Column(sa.String, nullable=False)
    author_id = sa.Column(sa.Integer, sa.ForeignKey("authors.id"))
    author = orm.relationship("Author",
                              backref=orm.backref("quotes", lazy="dynamic"))
    posted_at = sa.Column(sa.DateTime)

    def __init__(self, content, author):
        self.author = author
        self.content = content
        self.posted_at = datetime.utcnow()


# schema
class AuthorSchema(Schema):
    formatted_name = fields.Method("format_name")

    def format_name(self, author):
        return "{}, {}".format(author.last, author.first)

    class Meta:
        fields = ('id', 'first', 'last', "formatted_name")


def must_not_be_blank(data):
    if not data:
        raise ValidationError('Data not provided.')


class QuoteSchema(Schema):
    author = fields.Nested(AuthorSchema, validate=must_not_be_blank)
    content = fields.Str(required=True, validate=must_not_be_blank)

    class Meta:
        fields = ("id", "content", "posted_at", 'author')

author_schema = AuthorSchema()
quote_schema = QuoteSchema()
quotes_schema = QuoteSchema(many=True, only=('id', 'content'))


# http excpetion
class APIError(HTTPError):
    code = 400
    msg = "APIError"

    def __init__(self, msg=None):
        body = {'status': self.code, 'message': msg or self.msg}
        Response.__init__(self, json.dumps(body))
        self.status = self.code
        self.content_type = 'application/json'


class APIBadRequest(APIError):
    pass



# view
@view_config(route_name="authors", renderer="json")
def get_authors(context, request):
    authors = Author.query.all()
    # Serialize the queryset
    serializer = AuthorSchema(many=True)
    result = serializer.dump(authors)
    return {"authors": result.data}


@view_config(route_name="author", renderer="json")
def get_author(context, request):
    pk = request.matchdict["pk"]
    try:
        author = Author.query.get(pk)
    except IntegrityError:
        raise APIBadRequest({"message": "Author could not be found."})
    author_result = author_schema.dump(author)
    quotes_result = quotes_schema.dump(author.quotes.all())
    return {'author': author_result.data, 'quotes': quotes_result.data}


@view_config(route_name="quotes", renderer="json", request_method="GET")
def get_quotes(context, request):
    quotes = Quote.query.all()
    result = quotes_schema.dump(quotes)
    return {"quotes": result.data}


@view_config(route_name="quote", renderer="json")
def get_quote(context, request):
    pk = request.matchdict["pk"]
    try:
        quote = Quote.query.get(pk)
    except IntegrityError:
        raise APIBadRequest({"message": "Quote could not be found."})
    result = quote_schema.dump(quote)
    return {"quote": result.data}


@view_config(route_name="quotes", renderer="json", request_method="POST")
def new_quote(context, request):
    json_body = request.json_body
    author_name = json_body.get("author")
    if author_name:
        first, last = author_name.split(' ')
        author_input = dict(first=first, last=last)
    else:
        author_input = {}
    content_input = json_body.get('content')
    input_data = dict(author=author_input, content=content_input)
    # Validate the input data
    errors = quote_schema.validate(input_data)
    if errors:
        raise APIBadRequest(errors)
    author = Author.query.filter_by(first=first, last=last).first()
    if author is None:
        # Create a new author
        author = Author(first, last)
        Session.add(author)
    # Create new quote
    quote = Quote(content_input, author)
    Session.add(quote)
    Session.flush()
    result = quote_schema.dump(Quote.query.get(quote.id))
    return {"message": "Created new quote.",
            "quote": result.data}


def main():
    config = Configurator()

    engine = sa.create_engine('sqlite:///file.db', echo=True)
    Session.configure(bind=engine)
    Base.metadata.bind = engine
    Base.metadata.create_all()

    config.include("pyramid_translogger")
    config.include("pyramid_tm")

    config.add_route(name="authors", path="/api/v1/authors/")
    config.add_route(name="author", path="/api/v1/author/{pk}/")
    config.add_route(name="quotes", path="/api/v1/quotes/")
    config.add_route(name="quote", path="/api/v1/quote/{pk}/")

    config.scan(__name__)
    return config.make_wsgi_app()

if __name__ == "__main__":
    from waitress import serve
    app = main()
    serve(app, port=8080, host="0.0.0.0")