app

View Source

  1from flask import Flask, request, jsonify, make_response
  2from flask_limiter import Limiter
  3from flask_limiter.util import get_remote_address
  4from pydantic import BaseModel, Field, ValidationError
  5from typing import List
  6from cache_manager import invalidate_cache ,cache, cached_keys
  7import logging
  8
  9# Initialize Flask app and Limiter for rate limiting
 10app = Flask(__name__)
 11
 12limiter = Limiter(key_func=get_remote_address, app=app, default_limits=["5 per minute"])
 13
 14
 15# Initialize logging
 16logging.basicConfig(level=logging.DEBUG)
 17
 18
 19# Pydantic model to hold and validate employee information.
 20class Employee(BaseModel):
 21    id: int = Field(..., example=1)
 22    first_name: str = Field(..., example="John")
 23    last_name: str = Field(..., example="Doe")
 24    position: str = Field(..., example="Engineer")
 25
 26
 27# Data structure to hold employee data. Using a list for simplicity.
 28employees: List[Employee] = [
 29    Employee(id=1, first_name='John', last_name='Doe', position='Engineer'),
 30    Employee(id=2, first_name='Jane', last_name='Doe', position='Manager'),
 31    Employee(id=3, first_name='Mike', last_name='Johnson', position='Developer')
 32]
 33
 34
 35# Utility function to check if an employee ID already exists
 36def is_unique_id(employee_id: int) -> bool:
 37    for employee in employees:
 38        if employee.id == employee_id:
 39            return False
 40    return True
 41
 42@cache.memoize()
 43def get_employee_by_id(employee_id):
 44    # retrieve employee from database or data structure 
 45    employee = next((employee for employee in employees if employee.id == employee_id), None)
 46    return employee
 47
 48def after_employee_add_or_update():
 49    invalidate_cache("get_employee")
 50
 51
 52# ADD: Add a new employee
 53@app.route('/v1/employees', methods=['POST'])
 54@limiter.limit("4 per minute")
 55def add_employee():
 56    """
 57    Add a new employee.
 58    Validates the incoming data and ensures the employee ID is unique.
 59    Returns:
 60        JSON: New employee data if added successfully or an error message.
 61    """
 62    # Validate and construct an Employee object from JSON data
 63    try:
 64        new_employee = Employee(**request.json)
 65
 66    except ValidationError as e:
 67        return make_response(jsonify({"error": "Bad Request", "message": str(e)}), 400)
 68
 69    except Exception as e:
 70        return make_response(jsonify({"error": "Internal Server Error", "message": str(e)}), 500)
 71
 72    # Check for unique ID and return 409 if ID already exists
 73    if not is_unique_id(new_employee.id):
 74        return make_response(jsonify({"error": "Conflict", "message": "Employee ID already exists"}), 409)
 75    # Add the new employee
 76    employees.append(new_employee)
 77    # Return the new employee data with HTTP status code 201 (Created)
 78    return make_response(jsonify(new_employee.dict()), 201)
 79
 80
 81# READ: Get all employees
 82# Simple endpoint to read all data. No validation required.
 83@cache.cached(timeout=100, key_prefix='get_employees')  # cache for 1 minute
 84@app.route('/v1/employees', methods=['GET'])
 85@limiter.limit("30 per minute")
 86def get_employees():
 87    """
 88    Retrieve paginated existing employees.
 89    Supports pagination through 'page' and 'per_page' query parameters.
 90    Filters by position through the 'position' query parameter.
 91    Sorts by field through the 'sort_by' and 'order' query parameters.
 92    Returns:
 93        JSON: List of employee data on the current page or error message.
 94    """
 95    global employees
 96    try:
 97        # Get pagination parameters from query string
 98        page = int(request.args.get('page', 1))  # Default page is 1
 99        per_page = int(request.args.get('per_page', 10))  # Default 10 items per page
100        position_filter = request.args.get('position', None)  # Optional position filter
101        sort_by = request.args.get('sort_by', None)  # Optional sort by field
102        order = request.args.get('order', 'asc')  # Optional sort order
103
104        # filter by position if filter is provided
105        if position_filter is not None:
106            employees = [employee for employee in employees if employee.position == position_filter]
107        else:
108            employees = employees
109
110        # sort by field if sort_by is provided
111        if sort_by is not None:
112            employees = sorted(employees, key=lambda x: getattr(x, sort_by), reverse=order == 'desc')
113        else:
114            employees = employees
115        # Calculate start and end indices for the slice of data to return
116        start = (page - 1) * per_page
117        end = start + per_page
118
119        # Retrieve and paginate employee data, then convert to list of dicts
120        paginated_employees = employees[start:end]
121        meta_data = {
122            "total_employees": len(employees),
123            "page": page,
124            "per_page": per_page,
125        }
126
127        response = {
128            "data": [employee.dict() for employee in paginated_employees],
129            "meta": meta_data
130        }
131
132        return make_response(jsonify(response), 200)
133
134    except ValueError as e:
135        # ValueError: Invalid value for 'page' or 'per_page'
136        return make_response(jsonify({"error": "Bad Request", "message": f"Invalid pagination parameter: {e}"}), 400)
137
138    except AttributeError as e:
139        # AttributeError: One of the objects is missing the `dict()` method or similar
140        return make_response(jsonify({"error": "Internal Server Error", "message": f"An attribute error occurred: {e}"}), 500)
141
142    except Exception as e:
143        # General Exception: Something else went wrong
144        return make_response(jsonify({"error": "Internal Server Error", "message": f"An unexpected error occurred: {e}"}), 500)
145
146
147# UPDATE: Update an existing employee
148@app.route('/v1/employees/<int:employee_id>', methods=['PUT'])
149@limiter.limit("2 per minute")
150def update_employee(employee_id: int):
151    """
152    Update an existing employee by ID.
153    Parameters:
154        employee_id (int): Employee ID to update.
155    Returns:
156        JSON: Updated employee data or error message.
157    """
158    try:
159        # Find employee by ID
160        existing_employee = next((employee for employee in employees if employee.id == employee_id), None)
161        # Return 404 if employee not found
162        if existing_employee is None:
163            return make_response(jsonify({"error": "Not Found", "message": "Employee not found"}), 404)
164        # Create Employee object from JSON data
165        updated_employee = Employee(**request.json)
166        # Manually update existing employee data
167        existing_employee.first_name = updated_employee.first_name
168        existing_employee.last_name = updated_employee.last_name
169        existing_employee.position = updated_employee.position
170
171        # Invalidate cache or perform other actions
172        after_employee_add_or_update()
173        # Return updated data
174        return make_response(jsonify({"status": "success", "message": "Employee updated", "data": updated_employee.dict()}), 200)
175
176    except KeyError as e:
177        # KeyError: The JSON data is missing a necessary field
178        print(f"KeyError: {e}")
179        return make_response(jsonify({"error": "Bad Request", "message": f"Missing field {e}"}), 400)
180
181    except ValueError as e:
182        # ValueError: Data types are wrong or data is invalid in some other way
183        print(f"ValueError: {e}")
184        return make_response(jsonify({"error": "Bad Request", "message": "Invalid field value"}), 400)
185
186    except Exception as e:
187        # General Exception: Something else went wrong
188        print(f"An unexpected error occurred: {e}")
189        return make_response(jsonify({"error": "Internal Server Error", "message": "An unexpected error occurred"}), 500)
190
191
192# DELETE: Delete an existing employee
193@app.route('/v1/employees/<int:employee_id>', methods=['DELETE'])
194@limiter.limit("2 per minute")
195def delete_employee(employee_id):
196    """
197    Delete an employee by ID.
198    Verifies the employee exists before deletion and removes it from the list.
199    Returns:
200        JSON: A success message if deleted, or an error message if not found.
201    """
202    try:
203        # Find the existing employee by ID
204        existing_employee = next((employee for employee in employees if employee.id == employee_id), None)
205
206        # If the employee does not exist, return a 404 error
207        if existing_employee is None:
208            return make_response(jsonify({"error": "Not Found", "message": "Employee not found"}), 404)
209
210        # Remove the existing employee
211        employees.remove(existing_employee)
212
213        # Return a success message with HTTP status code 200 (OK)
214        return make_response(jsonify({"message": "Employee deleted"}), 200)
215    except Exception as e:
216        # Log the exception for debugging
217        print(f"An unexpected error occurred: {e}")
218
219        # Return a generic internal server error with HTTP status code 500
220        return make_response(jsonify({"error": "Internal Server Error", "message": f"An unexpected error occurred: {e}"}), 500)
221
222
223# Optional endpoint to get a single employee by ID
224@cache.memoize()
225@app.route('/v1/employees/<int:employee_id>', methods=['GET'])
226@limiter.limit("10 per minute")
227def get_employee(employee_id):
228    """
229    Retrieve an employee by their ID.
230    Parameters:
231        - employee_id (int): The ID of the employee to retrieve
232    Returns:
233        - JSON object containing the employee information and success status, or an error message if not found.
234    """
235    # Add the cache key for this specific employee
236    cache_key = f'get_employee_{employee_id}'
237    cached_keys.add(cache_key)
238    # Filter the list of employees to find the employee with the given ID
239    matching_employees = [employee.dict() for employee in employees if employee.id == employee_id]
240
241    # If no matching employee is found, return a 404 error
242    if len(matching_employees) == 0:
243        return make_response(jsonify({"status": "error", "message": "Employee not found"}), 404)
244    # Return a success message with HTTP status code 200 (OK)
245    return make_response(jsonify({"status": "success", "message": "Employee found", "data": matching_employees[0]}), 200)
246
247
248if __name__ == '__main__':
249    app.run(debug=True)

app = <Flask 'app'>

limiter = <flask_limiter.extension.Limiter object>

class Employee(pydantic.main.BaseModel): View Source

21class Employee(BaseModel):
22    id: int = Field(..., example=1)
23    first_name: str = Field(..., example="John")
24    last_name: str = Field(..., example="Doe")
25    position: str = Field(..., example="Engineer")

id: int

first_name: str

last_name: str

position: str

Inherited Members

pydantic.main.BaseModel: BaseModel; Config; dict; json; parse_obj; parse_raw; parse_file; from_orm; construct; copy; schema; schema_json; validate; update_forward_refs

employees: List[Employee] = [Employee(id=1, first_name='John', last_name='Doe', position='Engineer'), Employee(id=2, first_name='Jane', last_name='Doe', position='Manager'), Employee(id=3, first_name='Mike', last_name='Johnson', position='Developer')]

def is_unique_id(employee_id: int) -> bool: View Source

37def is_unique_id(employee_id: int) -> bool:
38    for employee in employees:
39        if employee.id == employee_id:
40            return False
41    return True

@cache.memoize()

def get_employee_by_id(employee_id): View Source

43@cache.memoize()
44def get_employee_by_id(employee_id):
45    # retrieve employee from database or data structure 
46    employee = next((employee for employee in employees if employee.id == employee_id), None)
47    return employee

def after_employee_add_or_update(): View Source

49def after_employee_add_or_update():
50    invalidate_cache("get_employee")

@app.route('/v1/employees', methods=['POST'])

@limiter.limit('4 per minute')

def add_employee(): View Source

54@app.route('/v1/employees', methods=['POST'])
55@limiter.limit("4 per minute")
56def add_employee():
57    """
58    Add a new employee.
59    Validates the incoming data and ensures the employee ID is unique.
60    Returns:
61        JSON: New employee data if added successfully or an error message.
62    """
63    # Validate and construct an Employee object from JSON data
64    try:
65        new_employee = Employee(**request.json)
66
67    except ValidationError as e:
68        return make_response(jsonify({"error": "Bad Request", "message": str(e)}), 400)
69
70    except Exception as e:
71        return make_response(jsonify({"error": "Internal Server Error", "message": str(e)}), 500)
72
73    # Check for unique ID and return 409 if ID already exists
74    if not is_unique_id(new_employee.id):
75        return make_response(jsonify({"error": "Conflict", "message": "Employee ID already exists"}), 409)
76    # Add the new employee
77    employees.append(new_employee)
78    # Return the new employee data with HTTP status code 201 (Created)
79    return make_response(jsonify(new_employee.dict()), 201)

Add a new employee. Validates the incoming data and ensures the employee ID is unique. Returns: JSON: New employee data if added successfully or an error message.

@cache.cached(timeout=100, key_prefix='get_employees')

@app.route('/v1/employees', methods=['GET'])

@limiter.limit('30 per minute')

def get_employees(): View Source

 84@cache.cached(timeout=100, key_prefix='get_employees')  # cache for 1 minute
 85@app.route('/v1/employees', methods=['GET'])
 86@limiter.limit("30 per minute")
 87def get_employees():
 88    """
 89    Retrieve paginated existing employees.
 90    Supports pagination through 'page' and 'per_page' query parameters.
 91    Filters by position through the 'position' query parameter.
 92    Sorts by field through the 'sort_by' and 'order' query parameters.
 93    Returns:
 94        JSON: List of employee data on the current page or error message.
 95    """
 96    global employees
 97    try:
 98        # Get pagination parameters from query string
 99        page = int(request.args.get('page', 1))  # Default page is 1
100        per_page = int(request.args.get('per_page', 10))  # Default 10 items per page
101        position_filter = request.args.get('position', None)  # Optional position filter
102        sort_by = request.args.get('sort_by', None)  # Optional sort by field
103        order = request.args.get('order', 'asc')  # Optional sort order
104
105        # filter by position if filter is provided
106        if position_filter is not None:
107            employees = [employee for employee in employees if employee.position == position_filter]
108        else:
109            employees = employees
110
111        # sort by field if sort_by is provided
112        if sort_by is not None:
113            employees = sorted(employees, key=lambda x: getattr(x, sort_by), reverse=order == 'desc')
114        else:
115            employees = employees
116        # Calculate start and end indices for the slice of data to return
117        start = (page - 1) * per_page
118        end = start + per_page
119
120        # Retrieve and paginate employee data, then convert to list of dicts
121        paginated_employees = employees[start:end]
122        meta_data = {
123            "total_employees": len(employees),
124            "page": page,
125            "per_page": per_page,
126        }
127
128        response = {
129            "data": [employee.dict() for employee in paginated_employees],
130            "meta": meta_data
131        }
132
133        return make_response(jsonify(response), 200)
134
135    except ValueError as e:
136        # ValueError: Invalid value for 'page' or 'per_page'
137        return make_response(jsonify({"error": "Bad Request", "message": f"Invalid pagination parameter: {e}"}), 400)
138
139    except AttributeError as e:
140        # AttributeError: One of the objects is missing the `dict()` method or similar
141        return make_response(jsonify({"error": "Internal Server Error", "message": f"An attribute error occurred: {e}"}), 500)
142
143    except Exception as e:
144        # General Exception: Something else went wrong
145        return make_response(jsonify({"error": "Internal Server Error", "message": f"An unexpected error occurred: {e}"}), 500)

Retrieve paginated existing employees. Supports pagination through 'page' and 'per_page' query parameters. Filters by position through the 'position' query parameter. Sorts by field through the 'sort_by' and 'order' query parameters. Returns: JSON: List of employee data on the current page or error message.

@app.route('/v1/employees/<int:employee_id>', methods=['PUT'])

@limiter.limit('2 per minute')

def update_employee(employee_id: int): View Source

149@app.route('/v1/employees/<int:employee_id>', methods=['PUT'])
150@limiter.limit("2 per minute")
151def update_employee(employee_id: int):
152    """
153    Update an existing employee by ID.
154    Parameters:
155        employee_id (int): Employee ID to update.
156    Returns:
157        JSON: Updated employee data or error message.
158    """
159    try:
160        # Find employee by ID
161        existing_employee = next((employee for employee in employees if employee.id == employee_id), None)
162        # Return 404 if employee not found
163        if existing_employee is None:
164            return make_response(jsonify({"error": "Not Found", "message": "Employee not found"}), 404)
165        # Create Employee object from JSON data
166        updated_employee = Employee(**request.json)
167        # Manually update existing employee data
168        existing_employee.first_name = updated_employee.first_name
169        existing_employee.last_name = updated_employee.last_name
170        existing_employee.position = updated_employee.position
171
172        # Invalidate cache or perform other actions
173        after_employee_add_or_update()
174        # Return updated data
175        return make_response(jsonify({"status": "success", "message": "Employee updated", "data": updated_employee.dict()}), 200)
176
177    except KeyError as e:
178        # KeyError: The JSON data is missing a necessary field
179        print(f"KeyError: {e}")
180        return make_response(jsonify({"error": "Bad Request", "message": f"Missing field {e}"}), 400)
181
182    except ValueError as e:
183        # ValueError: Data types are wrong or data is invalid in some other way
184        print(f"ValueError: {e}")
185        return make_response(jsonify({"error": "Bad Request", "message": "Invalid field value"}), 400)
186
187    except Exception as e:
188        # General Exception: Something else went wrong
189        print(f"An unexpected error occurred: {e}")
190        return make_response(jsonify({"error": "Internal Server Error", "message": "An unexpected error occurred"}), 500)

Update an existing employee by ID. Parameters: employee_id (int): Employee ID to update. Returns: JSON: Updated employee data or error message.

@app.route('/v1/employees/<int:employee_id>', methods=['DELETE'])

@limiter.limit('2 per minute')

def delete_employee(employee_id): View Source

194@app.route('/v1/employees/<int:employee_id>', methods=['DELETE'])
195@limiter.limit("2 per minute")
196def delete_employee(employee_id):
197    """
198    Delete an employee by ID.
199    Verifies the employee exists before deletion and removes it from the list.
200    Returns:
201        JSON: A success message if deleted, or an error message if not found.
202    """
203    try:
204        # Find the existing employee by ID
205        existing_employee = next((employee for employee in employees if employee.id == employee_id), None)
206
207        # If the employee does not exist, return a 404 error
208        if existing_employee is None:
209            return make_response(jsonify({"error": "Not Found", "message": "Employee not found"}), 404)
210
211        # Remove the existing employee
212        employees.remove(existing_employee)
213
214        # Return a success message with HTTP status code 200 (OK)
215        return make_response(jsonify({"message": "Employee deleted"}), 200)
216    except Exception as e:
217        # Log the exception for debugging
218        print(f"An unexpected error occurred: {e}")
219
220        # Return a generic internal server error with HTTP status code 500
221        return make_response(jsonify({"error": "Internal Server Error", "message": f"An unexpected error occurred: {e}"}), 500)

Delete an employee by ID. Verifies the employee exists before deletion and removes it from the list. Returns: JSON: A success message if deleted, or an error message if not found.

@cache.memoize()

@app.route('/v1/employees/<int:employee_id>', methods=['GET'])

@limiter.limit('10 per minute')

def get_employee(employee_id): View Source

225@cache.memoize()
226@app.route('/v1/employees/<int:employee_id>', methods=['GET'])
227@limiter.limit("10 per minute")
228def get_employee(employee_id):
229    """
230    Retrieve an employee by their ID.
231    Parameters:
232        - employee_id (int): The ID of the employee to retrieve
233    Returns:
234        - JSON object containing the employee information and success status, or an error message if not found.
235    """
236    # Add the cache key for this specific employee
237    cache_key = f'get_employee_{employee_id}'
238    cached_keys.add(cache_key)
239    # Filter the list of employees to find the employee with the given ID
240    matching_employees = [employee.dict() for employee in employees if employee.id == employee_id]
241
242    # If no matching employee is found, return a 404 error
243    if len(matching_employees) == 0:
244        return make_response(jsonify({"status": "error", "message": "Employee not found"}), 404)
245    # Return a success message with HTTP status code 200 (OK)
246    return make_response(jsonify({"status": "success", "message": "Employee found", "data": matching_employees[0]}), 200)

Retrieve an employee by their ID. Parameters: - employee_id (int): The ID of the employee to retrieve Returns: - JSON object containing the employee information and success status, or an error message if not found.