Models
Person
from dataclasses import dataclass
from typing import List
from flask_example import db
from models.Address import Address
@dataclass
class Person(db.Model):
id: int = db.Column(db.Integer, primary_key=True)
firstname: str = db.Column(db.String)
lastname = db.Column(db.String)
addresses = db.relationship('Address',
back_populates='person',
lazy='select',
cascade="all, delete-orphan")
# This constructor is not mandatory, but to have a check about parameter passed if we want to
# create manually a person instance, it is better to define it
# Beware: the parameters must be compliant with the marshmallow schema.
def __init__(self, firstname: str, lastname: str, id: int = None, addresses: List[str] = None):
print(f'Person CONSTRUCTOR')
self.id = id
self.lastname = lastname
self.firstname = firstname
if not addresses:
addresses = []
self.addresses = addresses
def add_address(self, address: Address):
self.addresses.append(address)
def __repr__(self) -> str:
return f'Person: id={self.id}, firstname={self.firstname}, lastname={self.lastname},' \
# f'addresses={self.addresses}' |
Address
from dataclasses import dataclass
from flask_example import db
@dataclass
class Address(db.Model):
id = db.Column(db.Integer, primary_key=True)
country_code: str = db.Column(db.String)
zip_code = db.Column(db.String)
street = db.Column(db.String)
person = db.relationship('Person', back_populates='addresses')
person_id = db.Column(db.Integer, db.ForeignKey('person.id'), nullable=False)
def __init__(self, country_code: str, zip_code: str, street: str):
print(f'Address CONSTRUCTOR')
self.zip_code = zip_code
self.country_code = country_code
self.street = street
def __repr__(self) -> str:
return f'Address: id={self.id}, country_code={self.country_code}, zip_code={self.zip_code}, ' \
f'street={self.street}' |
Marshmallow schemas: SQLAlchemyAutoSchema
Options class for SQLAlchemyAutoSchema:
The same options as SQLAlchemySchemaOpts, with the addition of:
– include_fk: Whether to include foreign fields; defaults to False.
– include_relationships: Whether to include relationships; defaults to False
Remarks on our configuration:
– Nested field allows to serialize/deserialize the relationship.
Because addresses is a one-to-many relationship, we need to specify many=true.
– These are default configurations to serialize/deserialize. We can override it at runtime for each usage.
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema
from marshmallow_sqlalchemy import fields
from models.Address import Address
from models.Person import Person
class AddressSchema(SQLAlchemyAutoSchema):
class Meta:
model = Address
include_fk = True
load_instance = True
class PersonSchema(SQLAlchemyAutoSchema):
class Meta:
model = Person
include_relationships = True
load_instance = True
addresses = fields.Nested(AddressSchema, many=True) |
REST json routes
Add a person and optionally his addresses
Some remarks:
– The person_id field of address is required according to the model, so we need to ignore this check during the deserialization.
– SQLAlchemyAutoSchema.load() needs either a session or to specify we want to work with a transient model.
Here we specify we want a transient model, because it spares some SQL queries.
@app.route("/api/person", methods=['POST'])
def add_person():
json_dic: Dict = request.json
person_controller_logger.info(f'json_dic input={json_dic}, type={type(json_dic)}')
# json_dic input={'lastname': 'David', 'firstname': 'boyoDoe', 'addresses': [{'street': '8 oxford
# strict', 'country_code': 'FR', 'zip_code': '75001'}]}, type=<class 'dict'>
person_schema: PersonSchema = PersonSchema(exclude=['addresses.person_id'])
person = person_schema.load(json_dic, transient=True)
person_controller_logger.info(f'person converted: {person}')
person_repository.save(person)
return jsonify({
"id": person.id
}) |
Update a person and replace his addresses
Some remarks:
– Because we want to override existing addresses, we don’t want to deserialize the
addresses.person_id.
@app.route("/api/person/<int:person_id>", methods=['PUT'])
def update_person(person_id: int):
person_controller_logger.info(f'update_person() with person_id={person_id}')
json_dic: Dict = request.json
person_controller_logger.info(f'json_dic input={json_dic}, type={type(json_dic)}')
person_schema: PersonSchema = PersonSchema(exclude=['addresses.person_id'])
person = person_schema.load(json_dic, transient=True)
person_controller_logger.info(f'person converted: {person}')
person_repository.save(person)
return jsonify({
"id": person.id
}) |
Get all persons with optional filters
Some remarks:
– request.args is a special dictionary type:
werkzeug.datastructures.ImmutableMultiDict that contains url query params.
– many=True: because we want to serialize multiple person instances.
– exclude=['addresses']: because we don’t want to retrieve addresses.
Without this attribute, SQL queries would be performed to retrieve addresses.
@app.route('/api/persons', methods=['GET'])
def get_persons():
print(f'request.args={request.args}, type={type(request.args)}')
country_code: str = request.args.get('country_code', None)
lastname: str = request.args.get('lastname', None)
person_controller_logger.info(f'get_persons() with lastname={lastname},'
f'country_code={country_code}')
persons: list[Person] = person_repository.find_all(lastname=lastname, country_code=country_code)
person_schema: PersonSchema = PersonSchema(many=True, exclude=['addresses'])
person_json: Dict = person_schema.dump(persons)
return person_json |
Find person by id with optional fetch addresses
Some remarks:
– If the person is not found in the database, we call: abort() that stops the execution of the function and raises a
werkzeug.exceptions.HTTPException
for the given status code.
To convert the exception into a REST json response, we can define an errorhandler that maps HTTPExceptions to json responses.
@app.route('/api/person/<int:person_id>', methods=['GET'])
def get_person(person_id: int):
print(f'request.args={request.args}, type={type(request.args)}')
# request.args=ImmutableMultiDict([('fetch_address', '')]), type=<class
# 'werkzeug.datastructures.ImmutableMultiDict'>
fetch_address: bool = True if 'fetch_address' in request.args else False
person_controller_logger.info(f'person_id input={person_id},fetch_address={fetch_address}')
person: Person = person_repository.find_by_id(person_id, fetch_address)
if not person:
abort(404)
print(f'person={person}, type={type(person)}')
# If the object is not existing, the process is aborted here and the function returns 404 response
person_schema: PersonSchema = \
PersonSchema() if fetch_address else PersonSchema(exclude=['addresses'])
person_json: Dict = person_schema.dump(person)
return person_json |
Errorhandler example:
from flask import json
from werkzeug.exceptions import HTTPException
from flask_example import app
@app.errorhandler(HTTPException)
def handle_exception(e):
"""Return JSON instead of HTML for HTTP errors."""
# start with the correct headers and status code from the error
response = e.get_response()
# replace the body with JSON
response.data = json.dumps({
"code": e.code,
"name": e.name,
"description": e.description,
})
response.content_type = "application/json"
return response |
Http requests
Add person with 2 addresses:
curl -H "Content-Type: application/json" -d '{"lastname":"David", "firstname":"boyoDoe", "addresses":[{"street":"8 oxford street","country_code":"UK",
"zip_code":"99999"},{"street":"12 Circus","country_code":"FR", "zip_code":"75001"}]}' localhost:5001/api/person
Add another person with 2 addresses:
curl -H "Content-Type: application/json" -d '{"lastname":"Jon", "firstname":"fullbar", "addresses":[{"street":"8 special street","country_code":"US",
"zip_code":"99999"},{"street":"12 Zoo","country_code":"FR", "zip_code":"75001"}]}' localhost:5001/api/person
Get person and his address by id:<br />curl -X GET localhost:5001/api/person/1
curl -X GET localhost:5001/api/person/1?fetch_address
get all persons:
curl -X GET localhost:5001/api/persons
get all persons by filtering on the country code of the address and the last name of the person :
curl -X GET "localhost:5001/api/persons?country_code=FR&lastname=Jon"
update person:
curl -X PUT -H "Content-Type: application/json" -d '{"id": 1, "lastname":"Davido", "firstname":"noename", "addresses":[{"street":"12 Google",
"country_code":"FR", "zip_code":"75001"}]}' localhost:5001/api/person/1 |
Ce contenu a été publié dans
Non classé. Vous pouvez le mettre en favoris avec
ce permalien.
