Exemplu de API REST cu CRUD în Python

Un API mic și complet

Este momentul să punem piesele laolaltă: un API pentru utilizatori cu cele cinci rute clasice —

POST /users creare · GET /users listare · GET /users/{id} citirea unuia · PUT /users/{id} înlocuire · DELETE /users/{id} ștergere.

Datele sunt păstrate într-un dicționar în memorie, așa că exemplul nu depinde de nimic altceva. (Un server real tratează cererile în paralel, așa că o versiune de producție ar proteja dicționarul cu un lacăt (lock) și probabil l-ar înlocui cu o bază de date — forma funcțiilor care tratează cererile nu s-ar schimba.)

Creare și listare

POST parsează JSON-ul primit, îi dă utilizatorului un id, îl salvează și răspunde cu 201 și înregistrarea creată — inclusiv id-ul, ca clientul să-l afle. GET îi returnează pe toți ca tablou JSON.

POST /users și GET /users

Două creări, apoi lista: fiecare POST răspunde cu 201 și utilizatorul salvat, iar GET-ul final îi arată pe amândoi — starea API-ului a crescut cu fiecare cerere.

Scrie un program Python care implementează partea de creare și listare a unui REST API pentru utilizatori, peste un dicționar în memorie: POST /users parsează JSON-ul, atribuie un id și răspunde cu 201 și utilizatorul creat; GET /users răspunde cu tabloul JSON al tuturor utilizatorilor. Pornește serverul pe un port liber, creează utilizatorii Ana și Ben, listează-i și afișează fiecare cod de stare și fiecare corp.

> Ieșire: program Python complet

Prompt

Scrie un program Python care implementează partea de creare și listare a unui REST API pentru utilizatori, peste un dicționar în memorie: POST /users parsează JSON-ul, atribuie un id și răspunde cu 201 și utilizatorul creat; GET /users răspunde cu tabloul JSON al tuturor utilizatorilor. Pornește serverul pe un port liber, creează utilizatorii Ana și Ben, listează-i și afișează fiecare cod de stare și fiecare corp.

Ieșire: program Python complet

import json
import threading
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

users = {}
next_id = 1

class UsersHandler(BaseHTTPRequestHandler):
  def do_POST(self):
    global next_id
    length = int(self.headers.get("Content-Length", 0))
    try:
      data = json.loads(self.rfile.read(length))
    except json.JSONDecodeError:
      self.reply(400, {"error": "invalid JSON"})
      return
    user = {"id": next_id, "name": data["name"]}
    users[next_id] = user
    next_id += 1
    self.reply(201, user)

  def do_GET(self):
    self.reply(200, [users[k] for k in sorted(users)])

  def reply(self, status, data):
    self.send_response(status)
    self.send_header("Content-Type", "application/json")
    self.end_headers()
    self.wfile.write((json.dumps(data) + "\n").encode())

  def log_message(self, fmt, *args):
    pass

server = ThreadingHTTPServer(("127.0.0.1", 0), UsersHandler)
base = f"http://127.0.0.1:{server.server_address[1]}"
threading.Thread(target=server.serve_forever, daemon=True).start()

for name in ["Ana", "Ben"]:
  payload = json.dumps({"name": name}).encode()
  req = urllib.request.Request(base + "/users", data=payload,
                               headers={"Content-Type": "application/json"})
  with urllib.request.urlopen(req) as resp:
    print("POST /users ->", resp.status, resp.read().decode().strip())

with urllib.request.urlopen(base + "/users") as resp:
  print("GET  /users ->", resp.status, resp.read().decode().strip())

server.shutdown()

Citire, înlocuire, ștergere

Rutele /users/{id} completează setul: citirea răspunde cu 200 sau 404, înlocuirea pune datele noi sub id-ul existent, ștergerea răspunde cu 204. După ștergere, același GET care funcționa înainte returnează corect 404 — resursa nu mai există.

GET, PUT și DELETE pe /users/{id}

Întreaga viață a unei resurse: o citim, o înlocuim (răspunsul conține noul nume), o ștergem și nu mai reușim s-o citim — fiecare pas cu codul de stare corect.

Scrie un program Python care implementează GET, PUT și DELETE pentru /users/{id} peste un dicționar în memorie care conține inițial un utilizator: GET răspunde cu 200 sau 404 și JSON, PUT înlocuiește utilizatorul păstrând id-ul, DELETE răspunde cu 204. Pornește serverul pe un port liber și parcurge întregul ciclu de viață — citire, înlocuire, ștergere, citire din nou — afișând pentru fiecare pas metoda, codul de stare și corpul.

> Ieșire: program Python complet

Prompt

Scrie un program Python care implementează GET, PUT și DELETE pentru /users/{id} peste un dicționar în memorie care conține inițial un utilizator: GET răspunde cu 200 sau 404 și JSON, PUT înlocuiește utilizatorul păstrând id-ul, DELETE răspunde cu 204. Pornește serverul pe un port liber și parcurge întregul ciclu de viață — citire, înlocuire, ștergere, citire din nou — afișând pentru fiecare pas metoda, codul de stare și corpul.

Ieșire: program Python complet

import json
import threading
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

users = {1: {"id": 1, "name": "Ana"}}

class UserHandler(BaseHTTPRequestHandler):
  def user_id(self):
    return int(self.path.rsplit("/", 1)[-1])   # „/users/1” -> 1

  def do_GET(self):
    user = users.get(self.user_id())
    if user is None:
      self.reply(404, {"error": "user not found"})
    else:
      self.reply(200, user)

  def do_PUT(self):
    length = int(self.headers.get("Content-Length", 0))
    data = json.loads(self.rfile.read(length))
    user = {"id": self.user_id(), "name": data["name"]}   # păstrăm id-ul
    users[self.user_id()] = user
    self.reply(200, user)

  def do_DELETE(self):
    users.pop(self.user_id(), None)
    self.send_response(204)
    self.end_headers()

  def reply(self, status, data):
    self.send_response(status)
    self.send_header("Content-Type", "application/json")
    self.end_headers()
    self.wfile.write((json.dumps(data) + "\n").encode())

  def log_message(self, fmt, *args):
    pass

server = ThreadingHTTPServer(("127.0.0.1", 0), UserHandler)
base = f"http://127.0.0.1:{server.server_address[1]}"
threading.Thread(target=server.serve_forever, daemon=True).start()

def call(method, path, payload=None):
  data = payload.encode() if payload else None
  req = urllib.request.Request(base + path, data=data, method=method,
                               headers={"Content-Type": "application/json"})
  try:
    with urllib.request.urlopen(req) as resp:
      status, body = resp.status, resp.read().decode()
  except urllib.error.HTTPError as err:
    status, body = err.code, err.read().decode()
  print(f"{method:<6} {path} -> {status} {body.strip()}".rstrip())

call("GET", "/users/1")
call("PUT", "/users/1", '{"name": "Ana Maria"}')
call("DELETE", "/users/1")
call("GET", "/users/1")

server.shutdown()