Routes

Modify main.py to add the CRUD routes for the Book:

Add CRUD operations

from datetime import datetime
import sqlite3

from flask import Flask, request, g, jsonify

app = Flask(__name__)
DATABASE = 'test.db'

def get_db():
    db = getattr(g, '_database', None)
    if db is None:
        db = sqlite3.connect(DATABASE)
        g._database = db
        cursor = db.cursor()
        cursor.execute('''CREATE TABLE IF NOT EXISTS books (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            author TEXT NOT NULL,
            published_at TEXT NOT NULL,
            description TEXT NOT NULL,
            isbn TEXT NOT NULL,
            total_pages INTEGER NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
          )''')
        cursor.close()
    return db

@app.teardown_appcontext
def close_connection(exception):
    db = getattr(g, '_database', None)
    if db is not None:
        db.close()

def execute_query(query, values=()):
    cursor = get_db().cursor()
    cursor.execute(query, values)
    result = cursor.fetchall()
    get_db().commit()
    cursor.close()
    return result

# Define a health endpoint handler, use `/health` or `/`
@app.route('/')
def health():
    # Return a simple response indicating the server is healthy
    return {"status": "ok"}

@app.route('/books', methods=['GET'])
def get_books():
    query = "SELECT * FROM books"
    books = execute_query(query)
    return books

@app.route('/books/<int:book_id>', methods=['GET'])
def get_book(book_id):
    query = "SELECT * FROM books WHERE id = ?"
    books = execute_query(query, (book_id,))
    if not books:
        return {"error": "Record not found"}, 404
    return jsonify(books[0])

@app.route('/books', methods=['POST'])
def create_book():
    b = request.get_json()
    query = "INSERT INTO books (title, author, published_at, description, isbn, total_pages, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
    now = datetime.now()
    values = (b['title'], b['author'], b['published_at'], b['description'], b['isbn'], b['total_pages'], now, now)
    execute_query(query, values)
    return b, 201

@app.route('/books/<int:book_id>', methods=['PUT'])
def update_book(book_id):
    b = request.get_json()
    query = "UPDATE books SET title=?, author=?, published_at=?, description=?, isbn=?, total_pages=?, updated_at=? WHERE id=?"
    values = (b['title'], b['author'], b['published_at'], b['description'], b['isbn'], b['total_pages'], datetime.now(), book_id)
    execute_query(query, values)
    return b

@app.route('/books/<int:book_id>', methods=['DELETE'])
def delete_book(book_id):
    query = "DELETE FROM books WHERE id=?"
    execute_query(query, (book_id,))
    return '', 204

At this point, we use a local SQLite¹ database for demo purpose.

sqlite3 is a built-in database module for Python that allows you to interact with SQLite databases. SQLite is a lightweight, serverless, self-contained SQL database engine.

Try with curl

Create a new book:

curl -X POST \
  http://localhost:5000/books \
  -H 'Content-Type: application/json' \
  -d '{
    "title": "Sample Book",
    "author": "John Doe",
    "published_at": "2023-01-01",
    "description": "A sample book description",
    "isbn": "1234567890",
    "total_pages": 200
}'

It should respond with this:

{
  "author": "John Doe",
  "description": "A sample book description",
  "isbn": "1234567890",
  "published_at": "2023-01-01",
  "title": "Sample Book",
  "total_pages": 200
}

Fetch a single book by ID:

curl -X GET http://localhost:5000/books/1

Result:

[
  1,
  "Sample Book",
  "John Doe",
  "2023-01-01",
  "A sample book description",
  "1234567890",
  200,
  "2024-03-05 21:20:39.938647",
  "2024-03-05 21:20:39.938647"
]

List all books:

curl -X GET http://localhost:5000/books

Result list:

[
  [
    1,
    "Sample Book",
    "John Doe",
    "2023-01-01",
    "A sample book description",
    "1234567890",
    200,
    "2024-03-05 21:20:39.938647",
    "2024-03-05 21:20:39.938647"
  ],
  [
    2,
    "Great Book",
    "Bob Smith",
    "2003-01-01",
    "A sample book description",
    "1234567890",
    200,
    "2024-03-05 21:29:11.470872",
    "2024-03-05 21:29:11.470872"
  ]
]

Update an existing book

curl -X PUT \
  http://localhost:5000/books/1 \
  -H 'Content-Type: application/json' \
  -d '{
    "title": "Updated Book Title",
    "author": "Jane Smith",
    "published_at": "2023-01-01",
    "description": "A new description",
    "isbn": "1234567890",
    "total_pages": 200
}'

Result:

{
  "author": "Jane Smith",
  "description": "A new description",
  "isbn": "1234567890",
  "published_at": "2023-01-01",
  "title": "Updated Book Title",
  "total_pages": 200
}

Delete an existing book:

curl -X DELETE http://localhost:5000/books/1

It returns code 204 for a sucessful deletion.

The REST api server has formed its basic shape now. Not bad!

SQLite: https://www.sqlite.org/index.html ↩

❮ Prev Next ❯