Merhabalar,
Bu arada, Flask, PostgreSQL, SQLAlchemy, Flask-SQLAlchemy kütüphaneleri kullanılarak oluşturulmuş, http request metodlarına uygun JSON cevaplar gönderen, ORM tabanlı basit bir RESTful API uygulaması örneği paylaşmak isterim.
SQLAlchemy ve Flask-SQLAlchemy’yi sqlite3 ile de kullanabilirsiniz ancak PostgreSQL’i kullanmaya başlayıp, sağladığı imkanları gördükten sonra tercih etmeye başlayacağınızı düşünüyorum. Ayrıca tablolara karmaşık JSON verileri de ekleyebilirsiniz.
Kodları paylaşmadan önce gerekli kütüphaneleri listeleyeyim:
python -m pip install flask psycopg2 Flask-SQLAlchemy
psycopg2’yi kullanabilmek için PostgreSQL’in yüklü olması gerektiğini hatırlatmak isterim. PostgreSQL’i kurduktan sonra bir kullanıcı adı, şifre, bir veritabanı ve bir tablo oluşturmanız gerekiyor.
Genellikle kurulumdan sonra aşağıdaki ifadeyle PostgreSQL konsoluna giriş yaparsınız:
psql -U postgres
Yukardaki ifadeden sonra cli uygulamasına giriş yaparsınız. Sonra da bir kullanıcı adı ve şifre oluşturun kendinize:
CREATE USER serhank WITH PASSWORD '123456' CREATEDB;
Sonra oluşturduğunuz kullanıcı adıyla oturum açın:
SET SESSION AUTHORIZATION 'serhank';
Sonra bir tane veritabanı oluşturun:
CREATE DATABASE test;
Sonra da bir tane tablo oluşturun:
CREATE TABLE IF NOT EXISTS test (id SERIAL PRIMARY KEY, data JSON);
Şimdi, PostgreSQL’in CLI uygulamasından çıkabiliriz.
Bu arada, ben bu kullanıcı adı, şifre, veritabanı ismi, host, port gibi bilgileri ortam değişkeni olarak saklıyorum.
Değerler de şöyle:
host = os.environ["POSTGRES_HOST"]
port = os.environ["POSTGRES_PORT"]
username = os.environ["POSTGRES_USERNAME"]
password = os.environ["POSTGRES_PASSWORD"]
database = os.environ["POSTGRES_DATABASE"]
Kodlarda yer alan bu ifadeleri kendinize göre değiştirirsiniz.
Şimdi izninizle dosyaları paylaşayım:
app.py
import os
import json
from sqlalchemy import and_
from models import db, Model
from flask import Flask, request, jsonify
host = os.environ["POSTGRES_HOST"]
port = os.environ["POSTGRES_PORT"]
username = os.environ["POSTGRES_USERNAME"]
password = os.environ["POSTGRES_PASSWORD"]
database = "test"
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = f"postgresql://{username}:{password}@{host}:{port}/{database}"
with app.app_context():
db.init_app(app)
db.drop_all()
db.create_all()
@app.route("/", methods=["GET"])
def get():
args = json.loads(request.data.decode("utf-8"))
if all(arg in ["category", "name"] and isinstance(args[arg], str) for arg in args):
query = Model.query.filter(and_(*map(lambda arg: Model.data[arg].astext == args[arg], args))).all()
if query:
return jsonify([i.data for i in query])
else:
return {"status": 404, "message": f"Not Found: {args['category']} not found"}
return {"status": 400, "message": "Bad Request"}
@app.route("/", methods=["POST"])
def post():
args = json.loads(request.data.decode("utf-8"))
if all(arg in ["name", "category"] and isinstance(args[arg], str) for arg in args):
instance = Model.query.filter(Model.data["name"].astext == args["name"]).first()
if not instance:
try:
db.session.add(Model(data=args))
db.session.commit()
return {"status": 201}
except Exception as e:
db.session.rollback()
return {"status": 500, "message": f"Internal Server Error: {e}"}
else:
return {"status": 409, "message": f"Conflict: {args['name']} already exists"}
return {"status": 400, "message": "Bad Request"}
@app.route("/", methods=["PUT"])
def put():
args = json.loads(request.data.decode("utf-8"))
if all(arg in ["name", "data"] for arg in args):
instance = Model.query.filter(Model.data["name"].astext == args["name"]).first()
if instance:
try:
db.session.query(Model).filter(Model.data["name"].astext == instance.data["name"]).update({"data": args["data"]})
db.session.commit()
return {"status": 200}
except Exception as e:
db.session.rollback()
return {"status": 500, "message": f"Internal Server Error: {e}"}
else:
try:
db.session.add(Model(data=args["data"]))
db.session.commit()
return {"status": 201}
except Exception as e:
db.session.rollback()
return {"status": 500, "message": f"Internal Server Error: {e}"}
return {"status": 400, "message": "Bad Request"}
@app.route("/", methods=["PATCH"])
def patch():
args = json.loads(request.data.decode("utf-8"))
if all(arg in ["name", "data"] for arg in args):
instance = Model.query.filter(Model.data["name"].astext == args["name"]).first()
if instance:
try:
db.session.query(Model).filter(Model.data["name"].astext == instance.data["name"]).update({"data": args["data"]})
db.session.commit()
return {"status": 200}
except Exception as e:
db.session.rollback()
return {"status": 500, "message": f"Internal Server Error: {e}"}
else:
return {"status": 404, "message": f"Not Found: {args['name']} not found"}
return {"status": 400, "message": "Bad Request"}
@app.route("/", methods=["DELETE"])
def delete():
args = json.loads(request.data.decode("utf-8"))
if list(args) == ["name"] and isinstance(args["name"], str):
instance = Model.query.filter(Model.data["name"].astext == args["name"]).first()
if instance:
try:
db.session.delete(instance)
db.session.commit()
return {"status": 204}
except Exception as e:
db.session.rollback()
return {"status": 500, "message": f"Internal Server Error: {e}"}
else:
return {"status": 404, "message": f"Not Found: {args['name']} not found"}
return {"status": 400, "message": "Bad Request"}
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=True)
model.py
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.dialects.postgresql import JSON
db = SQLAlchemy()
class Model(db.Model):
__tablename__ = "test"
id = db.Column(db.Integer, primary_key=True)
data = db.Column(JSON, nullable=False)
def __init__(self, data):
self.data = data
Server’ı çalıştırıp, başka bir dosyadan localhost’a http istek metodları gönderelim.
import json
import requests
post = requests.post("http://127.0.0.1:5000", data=json.dumps({"category": "sebze", "name": "patlıcan"}))
print(post.json())
post = requests.post("http://127.0.0.1:5000", data=json.dumps({"category": "sebze", "name": "pırasa"}))
print(post.json())
get = requests.get("http://127.0.0.1:5000", data=json.dumps({"category": "sebze"}))
print(get.json())
put = requests.put("http://127.0.0.1:5000", data=json.dumps({"name": "enginar", "data": {"category": "meyve", "name": "enginar"}}))
print(put.json())
patch = requests.patch("http://127.0.0.1:5000", data=json.dumps({"name": "enginar", "data": {"category": "sebze", "name": "enginar"}}))
print(patch.json())
delete = requests.delete("http://127.0.0.1:5000", data=json.dumps({"name": "enginar"}))
print(delete.json())
Çıktı:
{'status': 201}
{'status': 201}
[{'category': 'sebze', 'name': 'patlıcan'}, {'category': 'sebze', 'name': 'pırasa'}]
{'status': 201}
{'status': 200}
{'status': 204}
Umarım faydası olur. Herkese iyi çalışmalar.