328 أسطر
8.5 KiB
Python
328 أسطر
8.5 KiB
Python
from flask import Flask, request, jsonify, g
|
|
from flasgger import Swagger
|
|
import logging
|
|
import time
|
|
import threading
|
|
|
|
app = Flask(__name__)
|
|
swagger = Swagger(app)
|
|
|
|
# ---------- Logging setup ----------
|
|
logger = logging.getLogger("library_loans")
|
|
logger.setLevel(logging.INFO)
|
|
fh = logging.FileHandler("api.log")
|
|
formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
|
|
fh.setFormatter(formatter)
|
|
logger.addHandler(fh)
|
|
|
|
# ---------- In-memory storage ----------
|
|
# books: {id:int, title:str, author:str, year:int, copies:int}
|
|
# users: {id:int, name:str, email:str}
|
|
# loans: {id:int, user_id:int, book_id:int, timestamp:int, returned:bool}
|
|
books = []
|
|
users = []
|
|
loans = []
|
|
_next_loan_id = 1
|
|
_storage_lock = threading.Lock()
|
|
|
|
# ---------- request timing & logging ----------
|
|
@app.before_request
|
|
def start_timer():
|
|
g.start = time.time()
|
|
|
|
@app.after_request
|
|
def after_request(response):
|
|
latency = (time.time() - g.start) * 1000
|
|
client = request.remote_addr or "-"
|
|
msg = f"{client} {request.method} {request.path} {response.status_code} {latency:.2f}ms"
|
|
if 400 <= response.status_code < 600:
|
|
logger.error(msg)
|
|
else:
|
|
logger.info(msg)
|
|
return response
|
|
|
|
# ---------- Helpers ----------
|
|
def find_book(book_id):
|
|
return next((b for b in books if b["id"] == book_id), None)
|
|
|
|
def find_user(user_id):
|
|
return next((u for u in users if u["id"] == user_id), None)
|
|
|
|
def find_loan(loan_id):
|
|
return next((l for l in loans if l["id"] == loan_id), None)
|
|
|
|
def available_copies(book_id):
|
|
"""Return number of available (not loaned) copies for a book."""
|
|
total = 0
|
|
b = find_book(book_id)
|
|
if not b:
|
|
return 0
|
|
total = b.get("copies", 1)
|
|
borrowed_count = sum(1 for l in loans if l["book_id"] == book_id and not l.get("returned", False))
|
|
return total - borrowed_count
|
|
|
|
# ---------- Endpoints ----------
|
|
|
|
# 1) GET /books
|
|
@app.route("/books", methods=["GET"])
|
|
def get_books():
|
|
"""List all books
|
|
---
|
|
responses:
|
|
200:
|
|
description: list of books
|
|
"""
|
|
return jsonify(books), 200
|
|
|
|
# 2) POST /books
|
|
@app.route("/books", methods=["POST"])
|
|
def add_book():
|
|
"""Add a new book (id, title, author, year, copies)
|
|
---
|
|
parameters:
|
|
- in: body
|
|
name: body
|
|
schema:
|
|
type: object
|
|
required: [id, title, author]
|
|
properties:
|
|
id:
|
|
type: integer
|
|
title:
|
|
type: string
|
|
author:
|
|
type: string
|
|
year:
|
|
type: integer
|
|
copies:
|
|
type: integer
|
|
responses:
|
|
201:
|
|
description: created book
|
|
400:
|
|
description: invalid payload
|
|
409:
|
|
description: duplicate id
|
|
"""
|
|
data = request.get_json() or {}
|
|
if "id" not in data or "title" not in data or "author" not in data:
|
|
return jsonify({"error": "id, title, author required"}), 400
|
|
if find_book(data["id"]):
|
|
return jsonify({"error": "book id exists"}), 409
|
|
data.setdefault("year", None)
|
|
data.setdefault("copies", 1)
|
|
books.append({
|
|
"id": int(data["id"]),
|
|
"title": data["title"],
|
|
"author": data["author"],
|
|
"year": data["year"],
|
|
"copies": int(data["copies"])
|
|
})
|
|
return jsonify(find_book(data["id"])), 201
|
|
|
|
# 3) PUT /books/<id>
|
|
@app.route("/books/<int:book_id>", methods=["PUT"])
|
|
def update_book(book_id):
|
|
"""Update a book by id
|
|
---
|
|
parameters:
|
|
- name: book_id
|
|
in: path
|
|
type: integer
|
|
required: true
|
|
- in: body
|
|
name: body
|
|
schema:
|
|
type: object
|
|
responses:
|
|
200:
|
|
description: updated
|
|
404:
|
|
description: not found
|
|
"""
|
|
b = find_book(book_id)
|
|
if not b:
|
|
return jsonify({"error": "book not found"}), 404
|
|
data = request.get_json() or {}
|
|
# Allow updating title, author, year, copies
|
|
for k in ("title", "author", "year", "copies"):
|
|
if k in data:
|
|
if k == "copies":
|
|
b[k] = int(data[k])
|
|
else:
|
|
b[k] = data[k]
|
|
return jsonify(b), 200
|
|
|
|
# 4) DELETE /books/<id>
|
|
@app.route("/books/<int:book_id>", methods=["DELETE"])
|
|
def delete_book(book_id):
|
|
"""Delete book by id
|
|
---
|
|
parameters:
|
|
- name: book_id
|
|
in: path
|
|
type: integer
|
|
required: true
|
|
responses:
|
|
200:
|
|
description: deleted
|
|
404:
|
|
description: not found
|
|
"""
|
|
global books
|
|
if not find_book(book_id):
|
|
return jsonify({"error": "book not found"}), 404
|
|
books = [b for b in books if b["id"] != book_id]
|
|
# optionally remove related loans? keep history (we keep loans)
|
|
return jsonify({"message": "book deleted"}), 200
|
|
|
|
# 5) GET /users
|
|
@app.route("/users", methods=["GET"])
|
|
def get_users():
|
|
"""List users
|
|
---
|
|
responses:
|
|
200:
|
|
description: list of users
|
|
"""
|
|
return jsonify(users), 200
|
|
|
|
# 6) POST /users
|
|
@app.route("/users", methods=["POST"])
|
|
def add_user():
|
|
"""Add a new user (id, name, email)
|
|
---
|
|
parameters:
|
|
- in: body
|
|
name: body
|
|
schema:
|
|
type: object
|
|
required: [id, name]
|
|
properties:
|
|
id:
|
|
type: integer
|
|
name:
|
|
type: string
|
|
email:
|
|
type: string
|
|
responses:
|
|
201:
|
|
description: created
|
|
409:
|
|
description: duplicate
|
|
"""
|
|
data = request.get_json() or {}
|
|
if "id" not in data or "name" not in data:
|
|
return jsonify({"error": "id and name required"}), 400
|
|
if find_user(data["id"]):
|
|
return jsonify({"error": "user id exists"}), 409
|
|
users.append({
|
|
"id": int(data["id"]),
|
|
"name": data["name"],
|
|
"email": data.get("email")
|
|
})
|
|
return jsonify(find_user(data["id"])), 201
|
|
|
|
# 7) POST /loans -> borrow a book
|
|
@app.route("/loans", methods=["POST"])
|
|
def create_loan():
|
|
"""Borrow a book (user_id, book_id)
|
|
---
|
|
parameters:
|
|
- in: body
|
|
name: body
|
|
schema:
|
|
type: object
|
|
required: [user_id, book_id]
|
|
properties:
|
|
user_id:
|
|
type: integer
|
|
book_id:
|
|
type: integer
|
|
responses:
|
|
201:
|
|
description: loan created
|
|
404:
|
|
description: user/book not found
|
|
409:
|
|
description: not available
|
|
"""
|
|
global _next_loan_id
|
|
data = request.get_json() or {}
|
|
user_id = data.get("user_id")
|
|
book_id = data.get("book_id")
|
|
if not find_user(user_id):
|
|
return jsonify({"error": "user not found"}), 404
|
|
if not find_book(book_id):
|
|
return jsonify({"error": "book not found"}), 404
|
|
if available_copies(book_id) <= 0:
|
|
return jsonify({"error": "no copies available"}), 409
|
|
with _storage_lock:
|
|
loan = {
|
|
"id": _next_loan_id,
|
|
"user_id": int(user_id),
|
|
"book_id": int(book_id),
|
|
"timestamp": int(time.time()),
|
|
"returned": False
|
|
}
|
|
loans.append(loan)
|
|
_next_loan_id += 1
|
|
return jsonify(loan), 201
|
|
|
|
# 8) GET /loans -> list loans (active/all via query)
|
|
@app.route("/loans", methods=["GET"])
|
|
def list_loans():
|
|
"""List loans
|
|
---
|
|
parameters:
|
|
- name: active
|
|
in: query
|
|
type: boolean
|
|
description: if true, return only non-returned loans
|
|
responses:
|
|
200:
|
|
description: loans
|
|
"""
|
|
active = request.args.get("active")
|
|
if active is not None and active.lower() in ("1","true","yes"):
|
|
res = [l for l in loans if not l.get("returned", False)]
|
|
else:
|
|
res = loans
|
|
return jsonify(res), 200
|
|
|
|
# 9) PUT /loans/<id>/return -> return a loan
|
|
@app.route("/loans/<int:loan_id>/return", methods=["PUT"])
|
|
def return_loan(loan_id):
|
|
"""Return a borrowed book (mark loan returned)
|
|
---
|
|
parameters:
|
|
- name: loan_id
|
|
in: path
|
|
type: integer
|
|
required: true
|
|
responses:
|
|
200:
|
|
description: returned
|
|
404:
|
|
description: loan not found
|
|
400:
|
|
description: already returned
|
|
"""
|
|
l = find_loan(loan_id)
|
|
if not l:
|
|
return jsonify({"error": "loan not found"}), 404
|
|
if l.get("returned"):
|
|
return jsonify({"error": "already returned"}), 400
|
|
l["returned"] = True
|
|
l["returned_timestamp"] = int(time.time())
|
|
return jsonify(l), 200
|
|
|
|
# 10) Health
|
|
@app.route("/health", methods=["GET"])
|
|
def health():
|
|
return jsonify({"status": "ok"}), 200
|
|
|
|
# ---------- Run ----------
|
|
if __name__ == "__main__":
|
|
app.run(host="127.0.0.1", port=5000, debug=true)
|