Flask-RESTful z SQLAlchemy

W poprzednim artykule stworzyliśmy API do zarządzania zużyciem paliwa, które używało słownika jako bazy danych. Największą wadą tego rozwiązania był brak trwałości danych. Aby nasza aplikacja była bardziej użyteczna, zamiast słownika Pythona, użyjemy bazy danych SQL. Do tego idealnym wyborem jest SQLAlchemy – narzędzie do pracy z bazami danych SQL i obiektowy mapper relacyjny (ORM) w Pythonie.

Wymagania

1. Zainstalowany Python 3.x
2. Edytor kodu według uznania (np. PyCharm, Visual Studio Code)

Konfuguracja projektu

Jeśli projekt z poprzedniego artykułu jest już skonfigurowany, można pominąć kroki 1-3. Powstały nowe zależności, więc krok 4 trzeba wykonać ponownie dla nowych modułów.

1. Stwórz nowy projekt

mkdir fuel-consumption-api
cd drivers-api

2. Stwórz wizualne środowisko

python3 -m venv flask_venv

3. Aktywuj środowisko venv

source flask_venv/bin/activate

4. Zainstaluj wymagane zależności

a. Ręczenie przy użyciu managera pakietów pip

pip install flask
pip install flask-restful
pip install flask-sqlalchemy
pip install mysql-connector-python

b. Alternatywnie, możesz utworzyć plik requirements.txt z zależnościami:

flask
flask-restful
flask-sqlalchemy
mysql-connector-python

i zainstalować je za pomocą polecenia:

pip install -r requirements.txt 

Konfiguracja połączenia z bazą danych

Kod konfiguracji

from flask import Flask
from flask_restful import Resource, Api, abort, reqparse
from flask_sqlalchemy import SQLAlchemy


app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///fuel.db'
db = SQLAlchemy(app)
api = Api(app)

Opis kodu

1. SQLAlchemy jest dostępne w pakiecie flask_sqlalchemy.

from flask_sqlalchemy import SQLAlchemy

2. Dla celów projektu używamy bazy danych SQLite. Należy ustawić klucz konfiguracyjny SQLALCHEMY_DATABASE_URI i utworzyć obiekt SQLAlchemy.

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///fuel.db'
db = SQLAlchemy(app)

Do zmiennej URI przypisujemy następujące informacje:

dialect:///absolute\\path\to\fuel.db

Podczas łączenia się z innym typem bazy danych URI będzie zawierało więcej informacji:

dialect+driver://username:password@host:port/database

Model Bazy Danych

SQLAlchemy posiada ORM (Object-Relational Mapper). Aby go użyć, należy utworzyć model bazy danych.

Kod

class FuelConsumptionRecord(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    odometer = db.Column(db.Float, nullable=False)
    fuelQuantity = db.Column(db.Float, nullable=False)

    def serialize(self):
        return {
            'id': self.id,
            'odometer': self.odometer,
            'fuelQuantity': self.fuelQuantity
        }

Opis kodu

1. Model bazy danych to obiekt Pythona, który dziedziczy z klasy SQLAlchemy.Model. Pola klasy reprezentują kolumny w bazie danych.

class FuelConsumptionRecord(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    odometer = db.Column(db.Float, nullable=False)
    fuelQuantity = db.Column(db.Float, nullable=False)

2. W klasie FuelConsumptionRecord zdefiniowana jest metoda serialize. Celem tej metody jest łatwe serializowanie danych z obiektu, aby można je było zwrócić jako JSON.

    def serialize(self):
        return {
            'id': self.id,
            'odometer': self.odometer,
            'fuelQuantity': self.fuelQuantity
        }

Inicjalizacja bazy danych

Połączenie z bazą danych jest skonfigurowane, model jest zdefiniowany. Teraz można utworzyć tabelę.

1. Na początku należy utworzyć tabelę:

from fuel_consumption_api import db
db.create_all()

2. Tworzenie przykładowych danych:

from fuel_consumption_api import FuelConsumptionRecord
first_record =  FuelConsumptionRecord(odometer=0, fuelQuantity=0.0)
second_record = FuelConsumptionRecord(odometer=100, fuelQuantity=12.5)
third_record = FuelConsumptionRecord(odometer=110, fuelQuantity=12.5)

3. Dodanie danych do bazy

db.session.add(first_record)
db.session.add(second_record)
db.session.add(third_record)
db.session.commit()

4. Weryfikacja czy dane zostały zapisane

FuelConsumptionRecord.query.all()

Implementacja metod pobierania danych

Kod

from flask import Flask, jsonify
from flask_restful import Resource, Api, abort, reqparse
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///fuel.db'
db = SQLAlchemy(app)
api = Api(app)


class FuelConsumptionRecord(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    odometer = db.Column(db.Float, nullable=False)
    fuelQuantity = db.Column(db.Float, nullable=False)

    def serialize(self):
        return {
            'id': self.id,
            'odometer': self.odometer,
            'fuelQuantity': self.fuelQuantity
        }


parser = reqparse.RequestParser(bundle_errors=True)
parser.add_argument('odometer', type=float, required=True, help="odometer is required parameter!")
parser.add_argument('fuelQuantity', type=float, required=True, help="fuelQuantity is required parameter!")


class FuelConsumptionList(Resource):
    def get(self):
        records = FuelConsumptionRecord.query.all()
        return [FuelConsumptionRecord.serialize(record) for record in records]


class FuelConsumption(Resource):
    def get(self, record_id):
        return FuelConsumptionRecord.serialize(
            FuelConsumptionRecord.query.filter_by(id=record_id)
                .first_or_404(description='Record with id={} is not available'.format(record_id)))


api.add_resource(FuelConsumptionList, '/recordList',
                                      '/')
api.add_resource(FuelConsumption, '/record/<record_id>')

if __name__ == '__main__':
    app.run(debug=True)

Opis kodu

1. Aby pobrać wszystkie rekordy zapisane w bazie danych, wystarczy użyć query.all() na klasie FuelConsumptionRecord.

records = FuelConsumptionRecord.query.all()

2. Powyższe zapytanie zwraca listę obiektów FuelConsumptionRecord. Aby zwrócić je użytkownikowi, obiekty te muszą zostać zserializowane.

return [FuelConsumptionRecord.serialize(record) for record in records]

3. Aby pobrać konkretny rekord na podstawie pola record_id, zapytanie musi być przefiltrowane.

return FuelConsumptionRecord.serialize(
    FuelConsumptionRecord.query
                         .filter_by(id=record_id)
                         .first_or_404(description='Record with id={} is not available'.format(record_id)))

Aby zwrócić tylko jeden element, wywoływana jest metoda first_or_404. Zaletą tej metody jest to, że w przypadku gdy zapytanie zwraca zero wyników, zwracany jest kod statusu 404 HTTP, zamiast wewnętrznego błędu serwera. Komunikat o błędzie można łatwo skonfigurować za pomocą parametru description.

Testowanie

Sprawdźmy, czy wymagania biznesowe dotyczące pobierania danych zostały spełnione.

1. Użytkownik może pobrać wszystkie rekordy

Zapytanie:

curl -X GET http://localhost:5000/recordList

Oczekiwana odpowiedź:

    [
        {
            "id": 1,
            "odometer": 0.0,
            "fuelQuantity": 0.0
        },
        {
            "id": 2,
            "odometer": 100.0,
            "fuelQuantity": 12.5
        },
        {
            "id": 3,
            "odometer": 110.0,
            "fuelQuantity": 12.5
        }
    ]

2. Użytkownik może pobrać pojedynczy rekord

Zapytanie:

curl -X GET http://localhost:5000/record/2

Oczekiwana odpowiedź:

    {
        "id": 2,
        "odometer": 100.0,
        "fuelQuantity": 12.5
    }

Impementacja metody do aktualizacji lub usunięcia rekordu

Kod

from flask import Flask, jsonify
from flask_restful import Resource, Api, abort, reqparse
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///fuel.db'
db = SQLAlchemy(app)
api = Api(app)


class FuelConsumptionRecord(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    odometer = db.Column(db.Float, nullable=False)
    fuelQuantity = db.Column(db.Float, nullable=False)

    def serialize(self):
        return {
            'id': self.id,
            'odometer': self.odometer,
            'fuelQuantity': self.fuelQuantity
        }


parser = reqparse.RequestParser(bundle_errors=True)
parser.add_argument('odometer', type=float, required=True, help="odometer is required parameter!")
parser.add_argument('fuelQuantity', type=float, required=True, help="fuelQuantity is required parameter!")


class FuelConsumptionList(Resource):
    def get(self):
        records = FuelConsumptionRecord.query.all()
        return jsonify([FuelConsumptionRecord.serialize(record) for record in records])

    def post(self):
        args = parser.parse_args()
        fuel_consumption_record = FuelConsumptionRecord(odometer=args['odometer'], fuelQuantity=args['fuelQuantity'])
        db.session.add(fuel_consumption_record)
        db.session.commit()
        return FuelConsumptionRecord.serialize(fuel_consumption_record), 201


class FuelConsumption(Resource):
    def get(self, record_id):
        return jsonify(FuelConsumptionRecord.serialize(
            FuelConsumptionRecord.query.filter_by(id=record_id)
                .first_or_404(description='Record with id={} is not available'.format(record_id))))

    def delete(self, record_id):
        record = FuelConsumptionRecord.query.filter_by(id=record_id)\
            .first_or_404(description='Record with id={} is not available'.format(record_id))
        db.session.delete(record)
        db.session.commit()
        return '', 204

    def put(self, record_id):
        args = parser.parse_args()
        record = FuelConsumptionRecord.query.filter_by(id=record_id)\
            .first_or_404(description='Record with id={} is not available'.format(record_id))
        record.odometer = args['odometer']
        record.fuelQuantity = args['fuelQuantity']
        db.session.commit()
        return FuelConsumptionRecord.serialize(record), 201

api.add_resource(FuelConsumptionList, '/recordList',
                                      '/')
api.add_resource(FuelConsumption, '/record/<record_id>')


if __name__ == '__main__':
    app.run(debug=True)

Opis kodu

1. Stworzenie nowego rekordu

    def post(self):
        args = parser.parse_args()
        fuel_consumption_record = FuelConsumptionRecord(odometer=args['odometer'], fuelQuantity=args['fuelQuantity'])
        db.session.add(fuel_consumption_record)
        db.session.commit()
        return FuelConsumptionRecord.serialize(fuel_consumption_record), 201

Aby utworzyć nowy rekord w bazie danych, została stworzona nowa metoda FuelConsumptionRecord. Parametry są odczytywane z ciała żądania. Nowo utworzony obiekt musi zostać dodany do sesji SQLAlchemy, a następnie zatwierdzony, aby zapisać go w bazie danych. Aby zwrócić dane nowo utworzonego obiektu w odpowiedzi, obiekt musi zostać zserializowany.

2. Aktualizacja istniejącego rekordu

    def put(self, record_id):
        args = parser.parse_args()
        record = FuelConsumptionRecord.query.filter_by(id=record_id)\
            .first_or_404(description='Record with id={} is not available'.format(record_id))
        record.odometer = args['odometer']
        record.fuelQuantity = args['fuelQuantity']
        db.session.commit()
        return FuelConsumptionRecord.serialize(record), 201

W przypadku aktualizacji rekordu, jako pierwszy krok należy pobrać obiekt z bazy danych. Następnie pola obiektu są nadpisywane, a sesja jest zatwierdzana.

3. Usunięcie rekordu

    def delete(self, record_id):
            record = FuelConsumptionRecord.query.filter_by(id=record_id)\
                .first_or_404(description='Record with id={} is not available'.format(record_id))
            db.session.delete(record)
            db.session.commit()
            return '', 204

Aby usunąć rekord z bazy danych, najpierw musimy pobrać obiekt z bazy danych, a następnie użyć metody delete z sesji SQLAlchemy, aby podjąć działanie. Na końcu sesja musi zostać zatwierdzona.

Testowanie

1. Użytkownik może zapisać rekord

Zapytanie:

    curl -X POST \
    http://localhost:5000/recordList \
    -H 'Content-Type: application/json' \
    -d '{
        "odometer": 400,
        "fuelQuantity": 12
    }'

Oczekiwana odpowiedź:

    {
        "id": 4,
        "odometer": 400.0,
        "fuelQuantity": 24.0
    }

Zwracana jest odpowiedź z kodem 201 (CREATED).

2. Użytkownik otrzymuje komunikat o błędzie po wysłaniu żądania bez wymaganych pól

Zapytanie:

    curl -X POST \
    http://localhost:5000/recordList \
    -H 'Content-Type: application/json' \
    -d '{}'

Oczekiwana odpowiedź:

Kod odpowiedzi HTTP 400 BAD REQUEST

    {
        "message": {
            "odometer": "odometer is required parameter!",
            "fuelQuantity": "fuelQuantity is required parameter!"
        }
    }

3. Użytkownik może zaktualizować rekord

Zapytanie:

    curl -X PUT \
    http://localhost:5000/record/3 \
    -H 'Content-Type: application/json' \
    -d '{
    "odometer": 120,
    "fuelQuantity": 13.7
    }'

Oczekiwana odpowiedź:

    {
        "id": 3,
        "odometer": 120.0,
        "fuelQuantity": 13.7
    }

4. Użytkownik może usunąć pojedynczy rekord

Zapytanie:

curl -X DELETE http://localhost:5000/record/2

Odpowiedź:

Zwracana jest odpowiedź z kodem 204 (NO_CONTENT).

5. Aby sprawdzić, czy powyższe operacje zakończyły się sukcesem, wywołaj metodę GET na endpoint /recordList.

Zapytanie:

curl -X GET http://localhost:5000/recordList

Oczekiwana odpowiedź:

Prawidłowa odpowiedź przedstawiona poniżej:

   [
        {
            "id": 1,
            "odometer": 0.0,
            "fuelQuantity": 0.0
        },
        {
            "id": 3,
            "odometer": 120.0,
            "fuelQuantity": 13.7
        },
        {
            "id": 4,
            "odometer": 400.0,
            "fuelQuantity": 24.0
        }
    ]

Implementacja metod statycznych

Kod

from flask import Flask, jsonify
from flask_restful import Resource, Api, abort, reqparse
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///fuel.db'
db = SQLAlchemy(app)
api = Api(app)


class FuelConsumptionRecord(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    odometer = db.Column(db.Float, nullable=False)
    fuelQuantity = db.Column(db.Float, nullable=False)

    def serialize(self):
        return {
            'id': self.id,
            'odometer': self.odometer,
            'fuelQuantity': self.fuelQuantity
        }


parser = reqparse.RequestParser(bundle_errors=True)
parser.add_argument('odometer', type=float, required=True, help="odometer is required parameter!")
parser.add_argument('fuelQuantity', type=float, required=True, help="fuelQuantity is required parameter!")


def calculate_consumption(fuel_quantity, distance):
    return fuel_quantity / distance * 100


def calculate_distance(start_distance, end_distance):
    return end_distance - start_distance


def get_record_by_order_desc(records_limit, record_order):
    return db.session.query(FuelConsumptionRecord).order_by(FuelConsumptionRecord.id.desc()).limit(records_limit)[record_order]


class FuelConsumptionList(Resource):
    def get(self):
        records = FuelConsumptionRecord.query.all()
        return jsonify([FuelConsumptionRecord.serialize(record) for record in records])

    def post(self):
        args = parser.parse_args()
        fuel_consumption_record = FuelConsumptionRecord(odometer=args['odometer'], fuelQuantity=args['fuelQuantity'])
        db.session.add(fuel_consumption_record)
        db.session.commit()
        return FuelConsumptionRecord.serialize(fuel_consumption_record), 201


class FuelConsumption(Resource):
    def get(self, record_id):
        return jsonify(FuelConsumptionRecord.serialize(
            FuelConsumptionRecord.query.filter_by(id=record_id)
                .first_or_404(description='Record with id={} is not available'.format(record_id))))

    def delete(self, record_id):
        record = FuelConsumptionRecord.query.filter_by(id=record_id)\
            .first_or_404(description='Record with id={} is not available'.format(record_id))
        db.session.delete(record)
        db.session.commit()
        return '', 204

    def put(self, record_id):
        args = parser.parse_args()
        record = FuelConsumptionRecord.query.filter_by(id=record_id)\
            .first_or_404(description='Record with id={} is not available'.format(record_id))
        record.odometer = args['odometer']
        record.fuelQuantity = args['fuelQuantity']
        db.session.commit()
        return FuelConsumptionRecord.serialize(record), 201


class LastFuelConsumption(Resource):
    def get(self):
        last_record = get_record_by_order_desc(1, 0)
        second_last_record = get_record_by_order_desc(2, 1)
        distance = calculate_distance(second_last_record.odometer, last_record.odometer)
        consumption = round(calculate_consumption(last_record.fuelQuantity, distance), 2)
        return {'lastFuelConsumption': consumption}


class AverageFuelConsumption(Resource):
    def get(self):
        records_count = FuelConsumptionRecord.query.count()
        sum_of_consumptions = 0
        for i in reversed(range(0, records_count)):
            start_record = get_record_by_order_desc(records_count, i-1)
            end_record = get_record_by_order_desc(records_count, i-2)
            distance = calculate_distance(start_record.odometer, end_record.odometer)
            sum_of_consumptions += calculate_consumption(end_record.fuelQuantity, distance)
        avg_consumption = round(sum_of_consumptions / (records_count - 1), 2)
        return {"avgFuelConsumption": avg_consumption}


api.add_resource(FuelConsumptionList, '/recordList',
                                      '/')
api.add_resource(FuelConsumption, '/record/<record_id>')
api.add_resource(LastFuelConsumption, '/calculateLastConsumption')
api.add_resource(AverageFuelConsumption, '/calculateAverageConsumption')

if __name__ == '__main__':
    app.run(debug=True)

Opis kodu:

Aby obliczyć ostatnie zużycie paliwa oraz średnie zużycie paliwa, używany jest kod z poprzedniego artykułu. Jedyną zmianą jest sposób pobierania danych z bazy danych.

1. get_record_by_order_desc to metoda pomocnicza do pobierania rekordów w kolejności malejącej według id.

def get_record_by_order_desc(records_limit, record_order):
    return db.session.query(FuelConsumptionRecord).order_by(FuelConsumptionRecord.id.desc()).limit(records_limit)[record_order]

Aby pobrać dane uporządkowane według pola, potrzebne jest inne podejście. W przeciwieństwie do poprzedniej implementacji, nie ma bezpośredniego zapytania na klasie FuelConsumptionRecord, lecz metody są wywoływane na obiekcie SQLAlchemy. Powodem tej zmiany jest możliwość użycia metody order_by.

Testowanie

1. Użytkownik może sprawdzić ostatnie zapisane zużycie paliwa

Zapytanie:

curl -X GET http://localhost:5000/calculateLastConsumption 

Oczekiwana odpowiedź:

    {
    "lastFuelConsumption": 8.57
    }

2. Użytkownik może sprawdzić średnie zużycie paliwa, oparte na wszystkich zapisanych rekordach

Zapytanie:

    curl -X GET http://localhost:5000/calculateAverageConsumption 

Oczekiwana odpowiedź:

    {
    "avgFuelConsumption": 9.99
    }

Podsumowanie

Flask-RESTful z SQLAlchemy to bardzo dobre połączenie do tworzenia rzeczywistych API. SQLAlchemy zapewnia mechanizmy do tworzenia złożonych zapytań oraz mapowania obiektów Python na dane w bazie danych. W tym artykule pokazałem, jak łatwy jest proces implementacji SQLAlchemy ORM w aplikacji Flask-RESTful.

Źródła:

• https://flask-sqlalchemy.palletsprojects.com/en/2.x/
• https://flask-restful.readthedocs.io/