# from flask import Flask import os import psycopg2 from dotenv import load_dotenv from flask_swagger_ui import get_swaggerui_blueprint import datetime import jwt import random import json import names from flask import ( Flask, jsonify, send_from_directory, request, ) from flask_cors import CORS, cross_origin from functools import wraps import json import psycopg2.extras import pandas as pd from flask_sqlalchemy import SQLAlchemy from sqlalchemy import create_engine # import socket #import SQLAlchemy #from werkzeug.utils import secure_filename load_dotenv() app = Flask(__name__) cors = CORS(app) # allow CORS for all domains on all routes. SWAGGER_URL = '/api/docs' # URL for exposing Swagger UI (without trailing '/') #API_URL = 'http://petstore.swagger.io/v2/swagger.json' # Our API url (can of course be a local resource) API_URL = 'http://localhost:6335/docs/digifi_swagger.json' # Call factory function to create our blueprint swaggerui_blueprint = get_swaggerui_blueprint( SWAGGER_URL, # Swagger UI static files will be mapped to '{SWAGGER_URL}/dist/'m API_URL, config={ # Swagger UI config overrides 'app_name': "digiFi-Core" }, # oauth_config={ # OAuth config. See https://github.com/swagger-api/swagger-ui#oauth2-configuration . # 'clientId': "your-client-id", # 'clientSecret': "your-client-secret-if-required", # 'realm': "your-realms", # 'appName': "your-app-name", # 'scopeSeparator': " ", # 'additionalQueryStringParams': {'test': "hello"} # } ) app.register_blueprint(swaggerui_blueprint) dataUrl = os.getenv("DATABASE_URL") connection = psycopg2.connect(dataUrl) @app.route('/') def hello(): return "Hello World!" @app.route('/salary/login', methods=["POST"]) def salary_login(): try: data = request.json if not data: return { "message": "Please provide user details", "data": None, "error": "Bad request" }, 400 # validate input username = data["username"] password = data["password"] if username == 'digifiuser' and password == 'digifipass' : return { "message": "Successfully Login", "data": [ {"uid":"425611f2-c692-4404-b93d-76ca7a5ce70","icon": "icon_user", "name": "Simulator User" , "status": '1', "id": 1} ] }, 201 else: return { "message": "Error fetching auth token!, invalid email or password", "data": None, "error": "Unauthorized" }, 404 except Exception as e: return { "message": "Something went wrong!", "error": str(e), "data": None }, 500 @app.route('/salary/verifypin', methods=["POST"]) def salary_verifypin(): try: data = request.json if not data: return { "message": "Please provide pin", "data": None, "error": "Bad request" }, 400 # validate input bvn = data["bvn"] pin = data["pin"] SELECT_ACC = "SELECT id, uid,pin,bvn FROM demo_bank_accounts WHERE bvn='" + bvn + "' AND pin = '" + pin + "' " with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SELECT_ACC) account = cursor.fetchall() account_found = count = len( account ) if account_found == 1 : return { "status": "1", "data": account }, 201 else: return { "status": "-1", "data": None, "message" : "Invalid Pin", "error": "1" }, 404 except Exception as e: return { "status": "0", "error": str(e), "message" : "Something went wrong", "data": None }, 500 @app.route('/salary/demousers') def salary_demousers(): dList = [] # VARCHAR(125), # offers INT DEFAULT 0, # INT DEFAULT 0, # current_loans INT DEFAULT 0, # VARCHAR(25), # VARCHAR(12), # VARCHAR(125), # VARCHAR(6), # SQL_INSERT = "INSERT INTO demo_bank_accounts (name, salary_account, mobile, bvn, email, pin) VALUES(%s,%s,%s,%s,%s,%s)" # for i in range(900): # new_data = ( # names.get_full_name() , # random.randint(0, 1), # '801-000-'+str( random.randint(1000, 9999) ) , # '8315000'+str( random.randint(1000, 9999) ), # 'demo+'+ names.get_first_name() +'@chiefsoft.net', # random.randint(1100, 9999) # ) # # print(new_data) # with connection: # with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: # cursor.execute(SQL_INSERT, new_data) # # print(names.get_full_name()) SELECT_DEMO_ENTRY = f"SELECT id,uid, name,offers,salary_account,current_loans,mobile,bvn, email, pin, added::text FROM demo_bank_accounts ORDER BY id ASC" # print(SELECT_DEMO_ENTRY) with connection: with connection.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: cursor.execute(SELECT_DEMO_ENTRY) select_demoS = cursor.fetchall() # print(select_demoS) demo_data = json.dumps( [dict(ix) for ix in select_demoS] ) # print(demo_data) # sample_range = random.randint(2, 5) # for x in range(sample_range): # calDate = datetime.datetime.utcnow() + datetime.timedelta(minutes=180 * random.randint(1, 20)) # new_l = { # "uid":"425611f2-c692-4404-b93d-76ca7a5ce7"+str(x), # "name": names.get_full_name() , # "added": calDate, # "offers":random.randint(1, 4), # "mobile": '801-000-'+str( random.randint(1000, 9999) ) , # "bvn": '8315000'+str( random.randint(1000, 9999) ), # "email": 'demo+'+ names.get_first_name() +'@chiefsoft.net' , # "current_loans" : [], # "pin": random.randint(1100, 9999) , # "salary_account": random.randint(0, 1) # } # dList.append(new_l) demo_data = { "last_update": datetime.datetime.utcnow(), "offers":products(), "list" : json.loads( demo_data) } return { "demo_data": demo_data, }, 200 # return jsonify(demo_data=demo_data) @app.route('/salary/products') def salary_products(): product_data = { "products": products() , "extra" : [] } return { "product_data": product_data, }, 200 @app.route('/salary/loanoffers') def salary_loanoffers(): offers_data = { "offers": offers() , "extra" : [] } return { "product_data": offers_data, }, 200 def products(): product_data = [ {"cid": "1", "description": "Product Loan 01" , "active" : 0 }, {"cid": "2", "description": "Product Loan 02" , "active" : 0 }, {"cid": "3", "description": "First Advance" , "active" : 1 }, ] return product_data def offers(): offers_data = [ {"cid": "425611f2-c692-4404-b93d-76ca7a5ce00", "description": "100,000 Naira for 30 Days" , "active" : 1 }, {"cid": "425611f2-c692-4404-b93d-76ca7a5ce01", "description": "300,000 Naira for 60 Days" , "active" : 1 }, {"cid": "425611f2-c692-4404-b93d-76ca7a5ce02", "description": "900,000 Naira for 90 Days" , "active" : 1 }, ] return offers_data def offers_detail(offer_id): offer_detail = [ {"cid": "425611f2-c692-4404-b93d-76ca7a5ce00", "description": "100,000 Naira for 30 Days" , "active" : 1 }, {"cid": "425611f2-c692-4404-b93d-76ca7a5ce01", "description": "300,000 Naira for 60 Days" , "active" : 1 }, {"cid": "425611f2-c692-4404-b93d-76ca7a5ce02", "description": "900,000 Naira for 90 Days" , "active" : 1 }, ] return offer_detail if __name__ == '__main__': app.run(host='0.0.0.0', port=8000)