Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Binary file added flask-sqlite/.coverage.keploy
Binary file not shown.
4 changes: 4 additions & 0 deletions flask-sqlite/.coveragerc
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
[run]
omit =
/usr/*
sigterm = true
35 changes: 35 additions & 0 deletions flask-sqlite/.gitignore
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
# Python bytecode
__pycache__/
*.py[cod]
*.pyo

# SQLite DB file (do not commit local dev DB)
*.db
instance/

# Keploy reports (tests are OK, reports are not needed)
keploy/reports/

# Coverage reports
coverage.*
*.cover
.hypothesis/

# Keploy traces/cache
*.log
keploy/test-set-*/coverage.*
keploy/test-set-*/mocks/

# Environment files
.env
.venv/
env/
venv/

# IDE files
.vscode/
.idea/

# macOS/Linux system files
.DS_Store
Thumbs.db
158 changes: 158 additions & 0 deletions flask-sqlite/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,158 @@
# Flask + SQLite Sample App with Keploy Integration

This is a simple **Student Management REST API** built using Python's Flask framework and SQLite for storage. It demonstrates basic **CRUD operations** (Create, Read, Update, Delete) and showcases how to write **API tests with [Keploy](https://keploy.io)** by auto-capturing HTTP calls as test cases.

---

## 🚀 Features

- 🧑 Add, retrieve, update, and delete students
- 💾 Uses SQLite — no external DB setup required
- 🔌 RESTful API with JSON input/output
- ✅ Auto-generate test cases using [Keploy CLI](https://docs.keploy.io)
- 🔄 Replay & validate API responses with Keploy

---

## 📦 Tech Stack

- **Flask** — Lightweight web framework
- **Flask-SQLAlchemy** — ORM for SQLite
- **SQLite** — Built-in relational DB
- **Keploy** — Testing toolkit for API auto-mocking and regression testing

---

## 🛠 Installation

1. **Clone the repository**
```bash
git clone https://github.com/<your-username>/samples-python.git
cd samples-python/flask-sqlite
```

2. Set up virtual environment (optional)
```bash
python3 -m venv venv
source venv/bin/activate
```

3. Install Dependencies

```bash
pip install -r requirements.txt
```
4. Run the Flask app

```bash
python app.py
```
---

## API Endpoints

```bash

| Method | Endpoint | Description |
| ------ | ---------------- | -------------------- |
| GET | `/students` | Get all students |
| POST | `/students` | Add a new student |
| PUT | `/students/<id>` | Update student by ID |
| DELETE | `/students/<id>` | Delete student by ID |

```
---

## Sample Curl Commands

http://localhost:5000/
```json
{
"message": "Flask Student API"
}
```


### Add a student

```bash
curl -X POST http://localhost:5000/students \
-H "Content-Type: application/json" \
-d '{"name": "Alice", "age": 21}'
```
# Get all students
curl http://localhost:5000/students



# Update student
curl -X PUT http://localhost:5000/students/1 \
-H "Content-Type: application/json" \
-d '{"name": "Alice Updated", "age": 22}'

# Delete student
curl -X DELETE http://localhost:5000/students/1

```
---

## Running Keploy Tests

Step 1: Record Tests
Start Keploy in record mode:

```bash
keploy record --command "python app.py"
```
Send some API requests via curl or Postman to generate test cases.

Step 2: Replay Tests
After recording, stop the app and run:

```bash
keploy test --command "python app.py"
```
> Keploy will replay the previously captured requests and validate responses.

## Folder Structure

```bash
flask-sqlite/
├── app.py # Flask app entry point
├── models.py # Student model
├── requirements.txt # Python dependencies
├── keploy.yml # Keploy config file
├── keploy/ # Auto-generated test cases
└── README.md # You are here!
```
---

## Contributing
> Want to improve or add another example (e.g. FastAPI, SQLModel, etc.)? Contributions are welcome! Fork this repo, create your example folder, and submit a PR.

## About Keploy
Keploy is a developer-friendly open-source testing toolkit that auto-generates test cases from API calls in real-time and replays them to catch regressions — without writing any test code.

> Built with ❤️ for the Open Source Community.

| Stage | Improvement Type | Description |
| ------- | ---------------------------- | ------------------------------------------- |
| Stage 1 | Validation & Pagination | Add search, filter, pagination for students |
| Stage 2 | Security | Add authentication, role-based access |
| Stage 3 | Code Quality | Integrate Marshmallow/Pydantic, add tests |
| Stage 4 | Documentation & Deployment | Add Swagger, Dockerize, CI/CD pipeline |
| Stage 5 | Frontend & Realtime Features | Build React UI, add WebSocket updates |

---

#### Stage 1

Get End points

/students?page=1&per_page=5 — Gets first 5 students.

/students?name=John — Filters students whose names contain "John".

/students?age=20 — Filters students aged 20.

/students?page=2&per_page=3&name=ann — Combination.
150 changes: 150 additions & 0 deletions flask-sqlite/app.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,150 @@
from flask import Flask, request, jsonify
from models import db, Student
from werkzeug.exceptions import HTTPException
import logging
import os
from dotenv import load_dotenv

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URI', 'sqlite:///students.db')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SECRET_KEY'] = os.getenv('SECRET_KEY', 'fallback_secret_key')
db.init_app(app)

# Config Logging
logging.basicConfig(level=logging.INFO)

def validate_student_data(data, partial=False):
if not data:
return "Missing JSON body"
if not partial:
if 'name' not in data:
return "Missing 'name'"
if 'age' not in data:
return "Missing 'age'"
if 'name' in data:
if not isinstance(data['name'], str) or not data['name'].strip():
return "Invalid 'name'"
if len(data['name'].strip()) > 100:
return "'name' too long"
if 'age' in data:
if not isinstance(data['age'], int):
return "Invalid 'age'"
if data['age'] < 0 or data['age'] > 150:
return "'age' out of valid range"
return None # No error

# Test Route
@app.route('/')
def home():
return jsonify({"success": True, "message": "Flask Student API"}), 200

# CRUD Routes
# Get all students
@app.route('/students', methods=['GET'])
def get_students():
try:
page = int(request.args.get('page',1))
per_page = int(request.args.get('per_page',10))
except ValueError:
return jsonify({"success":False, "error":"'page' and 'per_page' must be integers"}), 400

students = Student.query.all()


# Filtering by name partial match
name_filter = request.args.get('name')
if name_filter:
students_query = students_query.filter(Student.name.ilike(f'%{name_filter}%'))

# Filtering by age exact match
age_filter = request.args.get('age')
if age_filter and age_filter.isdigit():
students_query = students_query.filter(Student.age == int(age_filter))

pagination = students_query.paginate(page=page, per_page=per_page, error_out=False)
students = pagination.items

return jsonify({
"success": True,
"page": page,
"per_page": per_page,
"total": pagination.total,
"total_pages": pagination.pages,
"students":[
{"id": s.id, "name": s.name, "age": s.age} for s in students
]}), 200

# Add a new student
@app.route('/students', methods=['POST'])
def add_student():
try:
data = request.get_json(force=True)
except Exception:
return jsonify({"success": False, "error": "Invalid JSON"}), 400
error = validate_student_data(data)
if error:
return jsonify({"success": False, "error": error}), 400
student = Student(name=data['name'].strip(), age=data['age'])
db.session.add(student)
try:
db.session.commit()
except Exception as e:
db.session.rollback()
logging.error(f"DB Error: {e}")
return jsonify({"success": False, "error": "Database error"}), 500
return jsonify({"success": True, "id": student.id, "name": student.name}), 201

# Update a student
@app.route('/students/<int:id>', methods=['PUT'])
def update_student(id):
student = Student.query.get(id)
if not student:
return jsonify({"success": False, "error": "Student not found"}), 404
try:
data = request.get_json(force=True)
except Exception:
return jsonify({"success": False, "error": "Invalid JSON"}), 400
error = validate_student_data(data, partial=True)
if error:
return jsonify({"success": False, "error": error}), 400
if 'name' in data:
student.name = data['name'].strip()
if 'age' in data:
student.age = data['age']
try:
db.session.commit()
except Exception as e:
db.session.rollback()
logging.error(f"DB Error: {e}")
return jsonify({"success": False, "error": "Database error"}), 500
return jsonify({"success": True, "message": "Updated"}), 200

# Delete a student
@app.route('/students/<int:id>', methods=['DELETE'])
def delete_student(id):
student = Student.query.get(id)
if not student:
return jsonify({"success": False, "error": "Student not found"}), 404
try:
db.session.delete(student)
db.session.commit()
except Exception as e:
db.session.rollback()
logging.error(f"DB Error: {e}")
return jsonify({"success": False, "error": "Database error"}), 500
return jsonify({"success": True, "message": "Deleted"}), 200

# Global Exception Handler
@app.errorhandler(Exception)
def handle_exception(e):
# Handle HTTP errors
if isinstance(e, HTTPException):
return jsonify({"success": False, "error": str(e)}), e.code
logging.error(f"Unhandled Error: {e}")
return jsonify({"success": False, "error": "Internal server error"}), 500

if __name__ == '__main__':
with app.app_context():
db.create_all()
app.run() # Remove debug=True for production
Loading