hii

sample_project/app/__init__.py

@@ -0,0 +1,5 @@
+from config import Config
+from app.app_config import register_versions, app
+
+
+register_versions(Config.VERSIONS_ALLOWED)

sample_project/app/app_config.py

@@ -0,0 +1,20 @@
+from flask import Flask
+from app.v1.urls import v1
+
+
+app = Flask(__name__)
+
+
+VERSION_OBJECT_MAPPING = {
+    '1': v1,
+}
+
+
+def get_url_prefix(version):
+    return '/api/v{version_number}'.format(version_number=str(version))
+
+
+def register_versions(allowed_versions):
+    for version in allowed_versions:
+        app.register_blueprint(
+            VERSION_OBJECT_MAPPING[version], url_prefix=get_url_prefix(version))

sample_project/app/common_utils.py

@@ -0,0 +1,40 @@
+import json
+from flask import Response, request
+
+
+def render_success_response(response, msg='', status=1):
+    body = {
+        's': status,
+        'm': msg,
+        'd': response
+    }
+    return Response(json.dumps(body), status=200, content_type='application/json')
+
+
+def render_error_response(msg, status):
+    body = {
+        's': 0,
+        'm': msg,
+        'd': dict()
+    }
+    return Response(json.dumps(body), status=status, content_type='application/json')
+
+
+def extract_params():
+    if request.method == 'POST':
+        if request.content_type == 'application/x-www-form-urlencoded':
+            params = request.form
+        elif 'multipart/form-data' in request.content_type:
+            params = request
+        else:
+            params = request.get_json()
+        return params
+    else:
+        params = request.args
+        parsed_params = dict()
+        for key, val in dict(params).items():
+            if isinstance(val, list) and val:
+                parsed_params.update({key: val[0]})
+            else:
+                parsed_params.update({key: val})
+        return parsed_params

sample_project/app/connections.py

@@ -0,0 +1,75 @@
+from decimal import Decimal
+from datetime import datetime
+import pymysql as PYMYSQL
+
+
+class SqlConnection(object):
+
+    def __init__(self, db_config):
+        self.db_config = db_config
+        try:
+            self.connection = PYMYSQL.connect(**self.db_config)
+            self.cursor = self.connection.cursor(dictionary=True)
+        except Exception as e:
+            self.connection.close()
+
+    @staticmethod
+    def parsed_db_result(db_data) -> dict:
+        parsed_db_data = dict()
+        for key, val in db_data.items():
+            if isinstance(val, Decimal):
+                parsed_db_data.update({key: float(val)})
+            elif isinstance(val, datetime):
+                parsed_db_data.update({key: val.strftime('%Y-%m-%d %H:%M:%S')})
+            elif isinstance(val, str) and val == 'NULL':
+                parsed_db_data.update({key: None})
+            else:
+                parsed_db_data.update({key: val})
+        return parsed_db_data
+
+    def reconnect_db(self):
+        self.connection = PYMYSQL.connect(**self.db_config)
+        self.cursor = self.connection.cursor(dictionary=True)
+
+    def query_db(self, query, params=None) -> list:
+        try:
+            if params:
+                self.cursor.execute(query, params)
+            else:
+                self.cursor.execute(query)
+        except Exception as e:
+            self.reconnect_db()
+            self.cursor.execute(query)
+        result = self.cursor.fetchall()
+        return list(map(lambda row: self.parsed_db_result(row), result)) if result else list()
+
+    def query_db_one(self, query, params=None, parsed=True) -> dict:
+        try:
+            if params:
+                self.cursor.execute(query, params)
+            else:
+                self.cursor.execute(query)
+        except Exception as e:
+            self.reconnect_db()
+            self.cursor.execute(query)
+        result = self.cursor.fetchone()
+        if not result:
+            return dict()
+        return self.parsed_db_result(result) if parsed else result
+
+    def write_db(self, query, params=None) -> int:
+        try:
+            if params:
+                self.cursor.execute(query, params)
+            else:
+                self.cursor.execute(query)
+            self.connection.commit()
+        except Exception as e:
+            self.reconnect_db()
+            self.cursor.execute(query)
+            self.connection.commit()
+        return self.cursor.lastrowid
+
+    def __del__(self):
+        self.cursor.close()
+        self.connection.close()

sample_project/app/decorators.py

@@ -0,0 +1,17 @@
+from flask import request
+from functools import wraps
+from app.common_utils import render_error_response, extract_params
+
+
+def validate_request(func):
+
+    def extract_headers():
+        return dict(request.headers)
+
+    @wraps(func)
+    def decorated_function(*args, **kwargs):
+        params = extract_params()
+        headers = extract_headers()
+        return func(params=params, headers=headers, *args, **kwargs)
+
+    return decorated_function

sample_project/app/v1/urls.py

@@ -0,0 +1,32 @@
+from flask import Blueprint
+from app.v1.users import views as sample_subapp_views
+from app.v1.users.views1 import User_Book_API
+
+
+v1 = Blueprint('v1', __name__)
+
+
+# subapp1 urls
+sample_subapp_prefix = '/users'
+
+v1.add_url_rule(sample_subapp_prefix + '/getUserName', view_func=sample_subapp_views.GetUserName.as_view('endpoint_1'))
+
+
+v1.add_url_rule(
+    sample_subapp_prefix + '/getUserDetails',
+    view_func=sample_subapp_views.GetUserDetails.as_view('endpoint_2')
+)
+
+user_view = User_Book_API.as_view('user_book_api')
+
+v1.add_url_rule(sample_subapp_prefix + '/books', view_func=user_view, methods=['GET', "POST"])
+
+v1.add_url_rule(sample_subapp_prefix + '/books', view_func=user_view, methods=['PUT','DELETE'])
+
+v1.add_url_rule(
+    sample_subapp_prefix + '/books/<int:id>', 
+    view_func=user_view,
+    methods=['GET']
+)
+
+

sample_project/app/v1/users/service.py

@@ -0,0 +1,8 @@
+class SubApp1Service:
+
+    def __init__(self, params, headers):
+        self.params = params
+        self.headers = headers
+
+    def get_static_api_response(self):
+        return self.params, 'success'

sample_project/app/v1/users/views.py

@@ -0,0 +1,119 @@
+from flask.views import MethodView
+from app.decorators import validate_request
+from app.common_utils import render_success_response
+from app.v1.users.service import SubApp1Service
+from flask import jsonify
+from flask import request
+from app.connections import SqlConnection
+import pymysql
+
+
+
+def db_connection():
+    # conn=mysql.connector.connect(host='localhost',user="root",password="root",database= "demo")
+    # cur=conn.cursor()
+
+
+    db_config = {
+        'host': 'localhost',
+        'user': 'root',
+        'password': 'root',
+        'database': 'db'
+    }
+
+
+    connection = pymysql.connect(**db_config)
+    return(connection) 
+db_connection()
+class GetUserName(MethodView):
+
+    @validate_request
+    def get(self, params, headers, *args, **kwargs):
+        response, message = SubApp1Service(params, headers).get_static_api_response()
+        return render_success_response(response, message)
+
+class GetUserDetails(MethodView):
+    def get(self):
+       conn = db_connection()
+       cursor = conn.cursor(pymysql.cursors.DictCursor)
+       query = "SELECT * FROM employee"
+       cursor.execute(query)
+       results = cursor.fetchall()
+       cursor.close()
+       conn.close()
+       return jsonify({"message": "GET: Data received", "data": results})
+
+
+
+    @validate_request
+    def post(self, *args, **kwargs):
+        data = request.get_json()
+        id = data.get('id')
+        name = data.get('name')
+
+        conn = db_connection()
+        cursor = conn.cursor()
+
+        query = "INSERT INTO employee (id, name) VALUES (%s, %s)"
+        values = (id, name) 
+
+        cursor.execute(query, values)
+        conn.commit()
+
+        if cursor.rowcount == 0:
+            return jsonify({"message": "No user found with the provided ID"})
+
+        cursor.close()
+        conn.close()
+
+        return jsonify({"message": "User created successfully"})
+
+
+    @validate_request
+    def put(self, *args, **kwargs):
+        data = request.get_json()
+        id = data.get('id')
+        name = data.get('name')
+
+
+        conn = db_connection()
+        cursor = conn.cursor()
+
+        query = "UPDATE employee SET name=%s WHERE id=%s"
+        values = (name, id)  
+
+        cursor.execute(query, values)
+        conn.commit()
+
+        if cursor.rowcount == 0:
+            return jsonify({"message": "No user found with the provided ID"})
+
+        cursor.close()
+        conn.close()
+
+        return jsonify({"message": "User updated successfully"})
+
+
+    @validate_request
+    def delete(self, *args, **kwargs):
+        id = request.args.get('id') 
+
+        if not id:
+            return jsonify({"message": "ID is required"}), 400
+
+        conn = db_connection()
+        cursor = conn.cursor()
+
+        query = "DELETE FROM employee WHERE id = %s"
+        values = (id,)
+
+        cursor.execute(query, values)
+        conn.commit()
+
+        if cursor.rowcount == 0:
+            return jsonify({"message": "No user found with the provided ID"}), 404
+
+        cursor.close()
+        conn.close()
+
+        return jsonify({"message": "User deleted successfully"}), 200