Wie kann ich meine Python-Flask-API für eine bessere Leistung optimieren?

[avtar11]

Ich arbeite derzeit an einem Backend-Entwicklungsprojekt, um eine RESTful-API mit Python und Flask zu erstellen. Die Backend-Entwicklerseite von Scalers war für mich die primäre Quelle der Hilfe, doch mit zunehmender Menge an API-Abfragen habe ich eine Verschlechterung der Leistung beobachtet. Besonders in Zeiten hoher Auslastung sind die Reaktionszeiten langsamer, als ich es mir gewünscht hätte.

Hier ist eine vereinfachte Version meines Flask-API-Codes:

Code: Alles auswählen

from flask import Flask, jsonify

app = Flask(__name__)

# Sample data
data = [
    {"id": 1, "name": "Product A", "price": 10.99},
    {"id": 2, "name": "Product B", "price": 15.99},
    # More data entries...
]

@app.route('/api/products', methods=['GET'])
def get_all_products():
    return jsonify(data)

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

Ich glaube, dass ich Optimierungen vornehmen kann, um die Leistung meiner Flask-API zu verbessern, aber ich bin mir nicht sicher, wo ich anfangen soll. Könnte jemand bitte meinen Code überprüfen und Best Practices oder Änderungen vorschlagen, die mir helfen können, bessere Antwortzeiten zu erreichen, insbesondere wenn die Anzahl der API-Anfragen zunimmt? Danke schön!

[__deets__]

Wenn das alles ist, was du da machst (also letztlich einfach ein grosses, statisches Dictionary auszuliefern), dann gibt es eine ganz offensichtliche Verbesserung: benutz nginx und liefer das einfach als statische json-Datei aus. https://stackoverflow.com/questions/867 ... json-files

Und debug=True hat in einer Produktionsapp natuerlich auch nichts zu suchen.

[Sirius3]

@avtar11: ist hier wirklich der Python-Code das Bottleneck? Im Normalfall ist die Netzwerkanbindung schneller dicht, als dass Python nicht mehr hinterherkommt, einfache JSON-Antworten zu liefern.

[paddie]

Ist das wirklich nur ein statisches Dictionary? Oder kommen die Daten doch aus einer DB? Je nach Komplexität der Queries und Anbindung der DB kann das auch ganz schön ausbremsen...

[noisefloor]

Hallo,

und den eingebauten Server von Flask für Performancetests zu nutzen ist auch Quatsch. Der ist nur für die Entwicklung gedacht und läuft single threaded, d.h. der macht schön brav einen Request nach dem anderen.

Wenn du einen Lasttest machen willst (oder musst), dann solltest du mindestens einen WSGI-Applikationsserver nutzen, wie Gunicorn oder Waitress. Ist auch in der Flask-Doku beschrieben. Beide laufen in de Standardkonfig schon mal mit vier Threads. Bei Gunicorn kannst du bei Bedarf auch noch andere Worker einstellen, die ggf. zu deinem Anwendungsszenario besser passen.

Gruß, noisefloor

[DeaD_EyE]

Das Schlimmste, was man machen kann, sind Tests der Performance im Debug-Modus.

Um die Geschwindigkeit der Serialisierung zu testen, kann man sich Testdaten generieren.
Wenn deine API-Endpunkte weniger als 17576 Datensätze ausliefern, liegt das Problem nicht bei der Serialisierung.

Code: Alles auswählen

17576 Datensätze
json: 14.3 ms ± 203 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)
ujson: 10.2 ms ± 73.7 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)
sqlite3: 11.4 µs ± 552 ns per loop (mean ± std. dev. of 7 runs, 100,000 loops each)

(Intel(R) Core(TM) i5-7200U CPU @ 2.50GHz)

Das Ergebnis von sqlite3 ist wahrscheinlich verfälscht. Ich gehe von aus, dass dort ein Cache verwendet wird.

Code (ipython wird genutzt):

Code: Alles auswählen

import itertools
import string
import random
import json
import sqlite3
from pathlib import Path

import ujson

uni = lambda: random.uniform(0, 20)
data = [
    {"id": idx, "name": f"Name {''.join(name)}", "price": round(price, 2)}
    for idx, name, price in zip(
        itertools.count(1),
        itertools.product(*[string.ascii_uppercase] * 3),
        iter(uni, None),
    )
]

test_db = Path("test-db.sqlite")
test_db.unlink(missing_ok=True)
db = sqlite3.connect(test_db)
cur = db.cursor()
with db:
    cur.execute("CREATE TABLE data (id integer, name string, price float);")
    db.commit()
    cur.executemany("INSERT INTO data VALUES (:id, :name, :price);", data)


raw_data = json.dumps(data)
print(f"{len(data)} Datensätze")
print("json: ", end="")
get_ipython().run_line_magic('timeit', 'json.loads(raw_data)')
print("ujson: ", end="")
get_ipython().run_line_magic('timeit', 'ujson.loads(raw_data)')
print("sqlite3: ", end="")
get_ipython().run_line_magic('timeit', 'cur.execute("SELECT * FROM data;")')

Wenn deine API-Endpunkte viele Datenbank-Abfragen tätigen, solltest du dir über Caching Gedanken machen.

Du solltest auch jeden Fall ein Profiling der API-Endpunkte machen.
Wie das geht, wird z.B. hier beschrieben: https://codingshower.com/profiling-pyth ... -profiler/

Code: Alles auswählen

from flask import Flask
from werkzeug.middleware.profiler import ProfilerMiddleware


app = Flask(__name__)
app.wsgi_app = ProfilerMiddleware(app.wsgi_app)


@app.get("/")
def index():
    return "42"


if __name__ == "__main__":
    app.run()

Quelle: w.py
Mit flask:

Code: Alles auswählen

flask --app w run

Mit gunicorn:

Code: Alles auswählen

gunicorn w:app