PostgRESTはオープンソースのWebサーバで
PostgreSQLをRESTfulAPIに変換してくれます
これによりHTTPリクエストを送ることでDBと接続でき、簡単にCRUD処理を実装できます
https://postgrest.org/en/stable/tutorials/tut0.html より引用
先日、7月12日にバージョン11.0がリリースされました
PostgREST 公式サイト
チュートリアル
GitHub Releasesページ
サンプルコード
TODOリストのAPIのサンプルコードです
Pythonで作成しました
フレームワークにFlaskを使用しています
todo.py
import requests
from flask import Flask, jsonify, request
app = Flask(__name__)
# PostgRESTのURL
POSTGREST_URL = "http://localhost:3000" # 適切な値を設定してください
@app.route("/tasks", methods=["GET", "POST"])
def tasks():
if request.method == "GET":
response = requests.get(f"{POSTGREST_URL}/tasks")
return response.json()
elif request.method == "POST":
data = request.get_json()
response = requests.post(f"{POSTGREST_URL}/tasks", json=data)
return response.json(), response.status_code
@app.route("/tasks/<task_id>", methods=["GET", "PUT", "DELETE"])
def task_detail(task_id):
if request.method == "GET":
response = requests.get(f"{POSTGREST_URL}/tasks?id=eq.{task_id}")
task = response.json()
if len(task) == 0:
return jsonify({"message": "タスクは見つかりませんでした"}), 404
return jsonify(task[0])
elif request.method == "PUT":
data = request.get_json()
response = requests.put(
f"{POSTGREST_URL}/tasks?id=eq.{task_id}", json=data
)
return response.json(), response.status_code
elif request.method == "DELETE":
response = requests.delete(
f"{POSTGREST_URL}/tasks?id=eq.{task_id}"
)
return "", response.status_code
if __name__ == "__main__":
app.run()
- GET
タスクの取得、一覧の取得 - POST
タスクの登録 - PUT
タスクの更新 - DELETE
タスクの削除
/tasksのエンドポイントでタスクの一覧の取得、登録を行います
/tasks/＜task_id>のエンドポイントで個別のタスクの取得や、更新と削除を行います
curlだとこのようになります
/tasks - GET
curl http://localhost:3000/tasks
/tasks/＜task_id＞ - PUT
curl -X PUT -H "Content-Type: application/json" -d '{"status": completed}' http://localhost:3000/tasks/1
おまけ
Djangoを使用
Django REST Frameworkを使用
views.py
from django.shortcuts import get_object_or_404
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from tasks.models import Task
import json
@csrf_exempt
def task_list(request):
if request.method == "GET":
tasks = Task.objects.all()
data = [
{
"id": task.id,
"title": task.title,
"description": task.description,
"completed": task.completed
}for task in tasks]
return JsonResponse(data, safe=False)
elif request.method == "POST":
data = json.loads(request.body)
task = Task.objects.create(
title=data["title"],
description=data.get("description", ""),
completed=data.get("completed", False)
)
return JsonResponse(
{
"id": task.id,
"title": task.title,
"description": task.description,
"completed": task.completed
},
status=201
)
@csrf_exempt
def task_detail(request, task_id):
task = get_object_or_404(Task, id=task_id)
if request.method == "GET":
return JsonResponse(
{
"id": task.id,
"title": task.title,
"description": task.description,
"completed": task.completed
}
)
elif request.method == "PUT":
data = json.loads(request.body)
task.title = data.get("title", task.title)
task.description = data.get("description", task.description)
task.completed = data.get("completed", task.completed)
task.save()
return JsonResponse(
{
"id": task.id,
"title": task.title,
"description": task.description,
"completed": task.completed
}
)
elif request.method == "DELETE":
task.delete()
return JsonResponse({}, status=204)
FastAPIを使用
main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List
app = FastAPI()
class Task(BaseModel):
id: int
title: str
description: str = None
completed: bool
tasks = []
@app.get("/tasks/", response_model=List[Task])
async def get_tasks():
return tasks
@app.post("/tasks/", response_model=Task, status_code=201)
async def create_task(task: Task):
tasks.append(task)
return task
@app.get("/tasks/{task_id}", response_model=Task)
async def get_task(task_id: int):
for task in tasks:
if task.id == task_id:
return task
raise HTTPException(status_code=404, detail="タスクは見つかりませんでした")
@app.put("/tasks/{task_id}", response_model=Task)
async def update_task(task_id: int, updated_task: Task):
for index, task in enumerate(tasks):
if task.id == task_id:
tasks[index] = updated_task
return updated_task
raise HTTPException(status_code=404, detail="タスクは見つかりませんでした")
@app.delete("/tasks/{task_id}", response_model=None, status_code=204)
async def delete_task(task_id: int):
for index, task in enumerate(tasks):
if task.id == task_id:
del tasks[index]
return
raise HTTPException(status_code=404, detail="タスクは見つかりませんでした")
備考
ここでPostgRESTについて質問できます