Perplexity vs Qwen for Coding

Here's how to build a Python REST API endpoint with JSON validation, SQLite queries, pagination, and error handling using FastAPI and SQLAlchemy:[4]

Core Implementation

Database Setup and Models

from fastapi import FastAPI, Depends, HTTPException, Query
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel, Field
from typing import List

# Database configuration
DATABASE_URL = "sqlite:///./items.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# SQLAlchemy model
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String)

Base.metadata.create_all(bind=engine)

# Pydantic models for validation
class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: str = Field(..., min_length=1, max_length=500)

class ItemResponse(BaseModel):
    id: int
    name: str
    description: str

class PaginatedResponse(BaseModel):
    items: List[ItemResponse]
    total: int
    page: int
    page_size: int

Dependency for Database Session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

API Endpoints with Pagination and Error Handling

app = FastAPI()

# Create item
@app.post("/items/", response_model=ItemResponse, status_code=201)
async def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    try:
        db_item = Item(name=item.name, description=item.description)
        db.add(db_item)
        db.commit()
        db.refresh(db_item)
        return db_item
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=500, detail="Failed to create item")

# Read items with pagination
@app.get("/items/", response_model=PaginatedResponse)
async def read_items(
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    db: Session = Depends(get_db)
):
    try:
        total = db.query(Item).count()
        
        if page < 1:
            raise HTTPException(status_code=400, detail="Page must be >= 1")
        
        offset = (page - 1) * page_size
        items = db.query(Item).offset(offset).limit(page_size).all()
        
        return {
            "items": items,
            "total": total,
            "page": page,
            "page_size": page_size
        }
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail="Failed to retrieve items")

# Read single item
@app.get("/items/{item_id}", response_model=ItemResponse)
async def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

Key Features

Input Validation: Pydantic models automatically validate JSON payloads, enforcing type checking and field constraints (min/max length).[4]

Pagination: Query parameters (page and page_size) control result pagination, with validation to prevent invalid values. The response includes metadata (total count, current page, page size).

Error Handling: HTTP exceptions with appropriate status codes (400 for bad requests, 404 for not found, 500 for server errors) provide clear error messages. Database transactions use try-except blocks with rollback on failure.[4]

Database Operations: SQLAlchemy handles CRUD operations with session management through dependency injection, ensuring proper resource cleanup.[4]

Testing the Endpoint

Create items:

curl -X POST "http://127.0.0.1:8000/items/" \
  -H "Content-Type: application/json" \
  -d '{"name": "Item 1", "description": "A sample item"}'

Retrieve paginated results:

curl "http://127.0.0.1:8000/items/?page=1&page_size=5"

This implementation provides production-ready API development with comprehensive validation, error handling, and scalable pagination.[4]

Here's a complete, production-ready example of a Python REST API using Flask that accepts a JSON payload, validates input, queries a SQLite database, and returns paginated results with proper error handling.

---

📁 Project Structure

project/
├── app.py
├── database.py
├── models.py
├── schemas.py
├── requirements.txt
└── README.md

---

📦 requirements.txt

Flask==3.0.0
Flask-SQLAlchemy==3.1.1
marshmallow==3.20.1

Install dependencies:

pip install -r requirements.txt

---

🗃️ database.py

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

def init_db(app):
    db.init_app(app)
    with app.app_context():
        db.create_all()

---

🧱 models.py

from database import db
from datetime import datetime

class Item(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)
    description = db.Column(db.Text, nullable=True)
    price = db.Column(db.Float, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'description': self.description,
            'price': self.price,
            'created_at': self.created_at.isoformat()
        }

---

✅ schemas.py

from marshmallow import Schema, fields, validate, ValidationError

class ItemQuerySchema(Schema):
    page = fields.Int(load_default=1, validate=validate.Range(min=1))
    per_page = fields.Int(load_default=10, validate=validate.Range(min=1, max=100))
    search = fields.Str(load_default='')
    min_price = fields.Float(load_default=None)
    max_price = fields.Float(load_default=None)

class ItemCreateSchema(Schema):
    name = fields.Str(required=True, validate=validate.Length(min=1, max=100))
    description = fields.Str(load_default=None)
    price = fields.Float(required=True, validate=validate.Range(min=0))

---

🚀 app.py

from flask import Flask, request, jsonify
from database import db, init_db
from models import Item
from schemas import ItemQuerySchema, ItemCreateSchema
from marshmallow import ValidationError

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///items.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

init_db(app)

query_schema = ItemQuerySchema()
create_schema = ItemCreateSchema()

@app.route('/items', methods=['GET'])
def get_items():
    try:
        params = query_schema.load(request.args)
    except ValidationError as err:
        return jsonify({'error': 'Invalid query parameters', 'details': err.messages}), 400

    page = params['page']
    per_page = params['per_page']
    search = params['search']
    min_price = params['min_price']
    max_price = params['max_price']

    query = Item.query

    if search:
        query = query.filter(Item.name.ilike(f'%{search}%'))
    if min_price is not None:
        query = query.filter(Item.price >= min_price)
    if max_price is not None:
        query = query.filter(Item.price <= max_price)

    pagination = query.paginate(page=page, per_page=per_page, error_out=False)
    items = [item.to_dict() for item in pagination.items]

    return jsonify({
        'data': items,
        'pagination': {
            'page': pagination.page,
            'per_page': pagination.per_page,
            'total_items': pagination.total,
            'total_pages': pagination.pages,
            'has_next': pagination.has_next,
            'has_prev': pagination.has_prev
        }
    }), 200

@app.route('/items', methods=['POST'])
def create_item():
    try:
        data = create_schema.load(request.json or {})
    except ValidationError as err:
        return jsonify({'error': 'Invalid input', 'details': err.messages}), 400
    except Exception:
        return jsonify({'error': 'Invalid JSON'}), 400

    item = Item(**data)
    db.session.add(item)
    db.session.commit()

    return jsonify({'message': 'Item created', 'data': item.to_dict()}), 201

@app.errorhandler(404)
def not_found(error):
    return jsonify({'error': 'Resource not found'}), 404

@app.errorhandler(500)
def internal_error(error):
    db.session.rollback()
    return jsonify({'error': 'Internal server error'}), 500

if __name__ == '__main__':
    app.run(debug=True)

---

🧪 Example Usage

Create an Item

curl -X POST http://localhost:5000/items \
  -H "Content-Type: application/json" \
  -d '{"name": "Laptop", "description": "High performance", "price": 999.99}'

Query Items with Pagination

curl "http://localhost:5000/items?page=1&per_page=5&search=Laptop&min_price=500&max_price=1500"

---

✅ Features

• JSON input validation with Marshmallow
• SQLite database with SQLAlchemy ORM
• Paginated GET endpoint with filtering
• POST endpoint for creating items
• Comprehensive error handling
• Modular and production-ready structure

---

This implementation provides a solid foundation for building scalable REST APIs in Python. You can extend it with authentication, logging, Swagger docs, or Dockerization as needed.