update
This commit was merged in pull request #4.
This commit is contained in:
Azeez Muibi
+100
-96
@@ -1,114 +1,118 @@
|
||||
from flask import Blueprint, request, jsonify, send_from_directory
|
||||
from app.api.services import (
|
||||
AuthorizationService,
|
||||
TransactionService,
|
||||
LoanService,
|
||||
AuthService,
|
||||
DashboardService
|
||||
)
|
||||
from app.utils.logger import logger
|
||||
from app.api.middlewares import enforce_json, require_auth
|
||||
import os
|
||||
from flask_jwt_extended import (
|
||||
JWTManager,
|
||||
jwt_required,
|
||||
create_access_token,
|
||||
get_jwt_identity,
|
||||
create_refresh_token,
|
||||
)
|
||||
from flask import Blueprint, request, jsonify
|
||||
from app.api.services.loan import LoanService
|
||||
from app.api.services.transaction import TransactionService
|
||||
from app.api.services.auth_service import AuthService
|
||||
from app.api.services.dashboard_service import DashboardService
|
||||
from functools import wraps
|
||||
|
||||
api = Blueprint("api", __name__)
|
||||
api = Blueprint('api', __name__)
|
||||
|
||||
|
||||
@api.before_request
|
||||
def cors_middleware():
|
||||
"""Middleware applied globally to all API routes in this blueprint"""
|
||||
return enforce_json()
|
||||
# JWT Authentication decorator
|
||||
def token_required(f):
|
||||
@wraps(f)
|
||||
def decorated(*args, **kwargs):
|
||||
token = None
|
||||
|
||||
# Get token from header
|
||||
auth_header = request.headers.get('Authorization')
|
||||
if auth_header:
|
||||
if auth_header.startswith('Bearer '):
|
||||
token = auth_header.split(' ')[1]
|
||||
|
||||
if not token:
|
||||
return jsonify({'message': 'Token is missing!'}), 401
|
||||
|
||||
# Verify token
|
||||
payload = AuthService.verify_token(token)
|
||||
if not payload:
|
||||
return jsonify({'message': 'Token is invalid or expired!'}), 401
|
||||
|
||||
# Add user info to request
|
||||
request.user = payload
|
||||
|
||||
return f(*args, **kwargs)
|
||||
|
||||
return decorated
|
||||
|
||||
|
||||
# Swagger JSON file
|
||||
@api.route("/swagger.json", methods=["GET"])
|
||||
def swagger_json():
|
||||
swagger_dir = os.path.join("swagger")
|
||||
return send_from_directory(swagger_dir, "digifi_swagger.json")
|
||||
|
||||
|
||||
@api.route("/swagger/<path:filename>")
|
||||
def serve_paths(filename):
|
||||
swagger_dir = os.path.join("swagger")
|
||||
return send_from_directory(swagger_dir, filename)
|
||||
|
||||
|
||||
# Login endpoint
|
||||
@api.route("/login", methods=["POST"])
|
||||
@api.route('/login', methods=['POST'])
|
||||
def login():
|
||||
data = request.get_json()
|
||||
response = AuthService.login(data)
|
||||
return response
|
||||
|
||||
# Check if username and password are provided
|
||||
if not data or 'username' not in data or 'password' not in data:
|
||||
return jsonify({
|
||||
'error': 'Missing credentials',
|
||||
'message': 'Username and password are required'
|
||||
}), 400
|
||||
|
||||
username = data.get('username', '')
|
||||
password = data.get('password', '')
|
||||
|
||||
# Call the login method from AuthService
|
||||
result = AuthService.login(username, password)
|
||||
|
||||
# Check if result is a tuple (error response)
|
||||
if isinstance(result, tuple):
|
||||
return jsonify(result[0]), result[1]
|
||||
|
||||
return jsonify(result)
|
||||
|
||||
|
||||
# Dashboard endpoint
|
||||
@api.route("/dashboard", methods=["GET"])
|
||||
@jwt_required()
|
||||
@api.route('/dashboard', methods=['GET'])
|
||||
@token_required
|
||||
def get_dashboard():
|
||||
response = DashboardService.get_dashboard_data()
|
||||
return response
|
||||
# Call the dashboard service
|
||||
result = DashboardService.get_dashboard_data()
|
||||
return jsonify(result)
|
||||
|
||||
|
||||
# Get All Transactions Endpoint
|
||||
@api.route("/transactions", methods=["GET"])
|
||||
@jwt_required()
|
||||
def get_transactions():
|
||||
# Extract query parameters for filtering
|
||||
filters = {
|
||||
'account_id': request.args.get('account_id'),
|
||||
'type': request.args.get('type'),
|
||||
'channel': request.args.get('channel'),
|
||||
'start_date': request.args.get('start_date'),
|
||||
'end_date': request.args.get('end_date')
|
||||
}
|
||||
|
||||
# logger.info(f"Get transactions request received with filters: {filters}")
|
||||
response = TransactionService.process_request(filters)
|
||||
return response
|
||||
|
||||
|
||||
# Get All Loans Endpoint
|
||||
@api.route("/loans", methods=["GET"])
|
||||
@jwt_required()
|
||||
@api.route('/loans', methods=['GET'])
|
||||
@token_required
|
||||
def get_loans():
|
||||
# Extract query parameters for filtering
|
||||
filters = {
|
||||
'customer_id': request.args.get('customer_id'),
|
||||
'account_id': request.args.get('account_id'),
|
||||
'status': request.args.get('status'),
|
||||
'offer_id': request.args.get('offer_id'),
|
||||
'product_id': request.args.get('product_id'),
|
||||
'start_date': request.args.get('start_date'),
|
||||
'end_date': request.args.get('end_date'),
|
||||
'due_before': request.args.get('due_before'),
|
||||
'due_after': request.args.get('due_after')
|
||||
}
|
||||
# Extract query parameters
|
||||
customer_id = request.args.get('customer_id')
|
||||
loan_id = request.args.get('loan_id')
|
||||
status = request.args.get('status')
|
||||
offer_id = request.args.get('offer_id')
|
||||
product_id = request.args.get('product_id')
|
||||
start_date = request.args.get('start_date')
|
||||
end_date = request.args.get('end_date')
|
||||
|
||||
# logger.info(f"Get loans request received with filters: {filters}")
|
||||
response = LoanService.process_request(filters)
|
||||
return response
|
||||
# Call the loan service
|
||||
result = LoanService.process_request(
|
||||
customer_id=customer_id,
|
||||
loan_id=loan_id,
|
||||
status=status,
|
||||
offer_id=offer_id,
|
||||
product_id=product_id,
|
||||
start_date=start_date,
|
||||
end_date=end_date
|
||||
)
|
||||
|
||||
return jsonify(result)
|
||||
|
||||
|
||||
# Authorize endpoint
|
||||
@api.route("/Authorize", methods=["POST"])
|
||||
def authorize():
|
||||
data = request.get_json()
|
||||
# logger.info(f"Authorize request received: {data}")
|
||||
response = AuthorizationService.process_request(data)
|
||||
return response
|
||||
@api.route('/transactions', methods=['GET'])
|
||||
@token_required
|
||||
def get_transactions():
|
||||
# Extract query parameters
|
||||
account_id = request.args.get('account_id')
|
||||
transaction_id = request.args.get('transaction_id')
|
||||
type = request.args.get('type')
|
||||
channel = request.args.get('channel')
|
||||
start_date = request.args.get('start_date')
|
||||
end_date = request.args.get('end_date')
|
||||
|
||||
# Call the transaction service
|
||||
result = TransactionService.process_request(
|
||||
account_id=account_id,
|
||||
transaction_id=transaction_id,
|
||||
type=type,
|
||||
channel=channel,
|
||||
start_date=start_date,
|
||||
end_date=end_date
|
||||
)
|
||||
|
||||
# Authorize refresh endpoint
|
||||
@api.route("/AuthorizeRefresh", methods=["POST"])
|
||||
@jwt_required(refresh=True)
|
||||
def refresh():
|
||||
data = request.get_json()
|
||||
# logger.info(f"Authorize refresh request received: {data}")
|
||||
response = AuthorizationService.process_refresh_request()
|
||||
return response
|
||||
return jsonify(result)
|
||||
@@ -1,58 +1,66 @@
|
||||
from flask import jsonify
|
||||
from app.utils.logger import logger
|
||||
from app.api.services.base_service import BaseService
|
||||
from app.models.user import User
|
||||
from flask_jwt_extended import create_access_token
|
||||
from datetime import timedelta
|
||||
import jwt
|
||||
import datetime
|
||||
from flask import current_app
|
||||
|
||||
|
||||
class AuthService(BaseService):
|
||||
class AuthService:
|
||||
@staticmethod
|
||||
def login(data):
|
||||
def login(username, password):
|
||||
"""
|
||||
Process the login request.
|
||||
|
||||
Args:
|
||||
data (dict): Login credentials including username and password.
|
||||
|
||||
Returns:
|
||||
dict: A standardized response with JWT token and user information.
|
||||
Login method that checks for specific credentials and returns a JWT token
|
||||
"""
|
||||
try:
|
||||
# Extract credentials
|
||||
username = data.get('username')
|
||||
password = data.get('password')
|
||||
# Define valid credentials for testing
|
||||
valid_credentials = {
|
||||
"digifiuser": "digifipass",
|
||||
"admin": "admin123",
|
||||
"test": "test123"
|
||||
}
|
||||
|
||||
# Validate input
|
||||
if not username or not password:
|
||||
return jsonify({
|
||||
"message": "Username and password are required"
|
||||
}), 400
|
||||
|
||||
# Get user by username
|
||||
user = User.get_user_by_username(username)
|
||||
|
||||
# Check if user exists and password is correct
|
||||
if not user or not user.check_password(password):
|
||||
return jsonify({
|
||||
"message": "Invalid username or password"
|
||||
}), 401
|
||||
|
||||
# Create JWT token with 15 minute expiration
|
||||
access_token = create_access_token(
|
||||
identity=user.username,
|
||||
expires_delta=timedelta(minutes=15),
|
||||
additional_claims={"name": user.name}
|
||||
)
|
||||
|
||||
# Return token and user information
|
||||
return {
|
||||
"jwt_token": access_token,
|
||||
"name": user.name
|
||||
# Check if the provided credentials are valid
|
||||
if username in valid_credentials and password == valid_credentials[username]:
|
||||
# Generate JWT token with 15 minutes expiration
|
||||
payload = {
|
||||
'sub': username, # Subject (typically user ID)
|
||||
'iat': datetime.datetime.utcnow(), # Issued at
|
||||
'exp': datetime.datetime.utcnow() + datetime.timedelta(minutes=15), # Expiration (15 minutes)
|
||||
'role': 'admin' if username == 'admin' else 'user' # Role based on username
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"An error occurred during login: {str(e)}", exc_info=True)
|
||||
return jsonify({
|
||||
"message": "Internal Server Error"
|
||||
}), 500
|
||||
# Get the secret key from config
|
||||
secret_key = current_app.config.get('JWT_SECRET_KEY', 'default-secret-key')
|
||||
|
||||
# Generate the token
|
||||
token = jwt.encode(payload, secret_key, algorithm='HS256')
|
||||
|
||||
# Return the token and user info
|
||||
return {
|
||||
'jwt_token': token,
|
||||
'user': {
|
||||
'username': username,
|
||||
'role': 'admin' if username == 'admin' else 'user'
|
||||
},
|
||||
'expires_in': 900 # 15 minutes in seconds
|
||||
}
|
||||
else:
|
||||
# Return error for invalid credentials
|
||||
return {
|
||||
'error': 'Invalid credentials',
|
||||
'message': 'The username or password is incorrect'
|
||||
}, 401
|
||||
|
||||
@staticmethod
|
||||
def verify_token(token):
|
||||
"""
|
||||
Verify the JWT token
|
||||
"""
|
||||
try:
|
||||
# Get the secret key from config
|
||||
secret_key = current_app.config.get('JWT_SECRET_KEY', 'default-secret-key')
|
||||
|
||||
# Decode the token
|
||||
payload = jwt.decode(token, secret_key, algorithms=['HS256'])
|
||||
return payload
|
||||
except jwt.ExpiredSignatureError:
|
||||
return None # Token has expired
|
||||
except jwt.InvalidTokenError:
|
||||
return None # Invalid token
|
||||
Reference in New Issue
Block a user