khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20241228) Payments module revamped!

Browse Source
This commit is contained in:
khushal
2024-12-28 16:34:57 +05:30
parent 7ddae1707f
commit 5c17eb28ae
18 changed files with 1477 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/blueprints/finstitutions/payments/auth.py → api/blueprints/finstitutions/payments/auth_v2.py
+7 -4
View File
@@ -6,7 +6,7 @@
DATE:
Friday, 13th Dec., 2024
Saturday, 28th Dec., 2024
OBJECTIVE:
@@ -170,9 +170,9 @@ async def authorize_payment_gateway(
if inbound_data.client == "safaricomMPesaExpress":
success = await current_app.payment_controller.set_token_direct(
db_conn = current_app.sql_writer,
mongo_conn = current_app.data_mongo,
success = await current_app.payments_controller.set_token_direct(
sql_conn = current_app.sql_writer,
mongo_data_conn = current_app.data_mongo,
auth_token = CoreAuthTokenModel(
serviceType = "paymentGateway",
client = inbound_data.client,
@@ -185,6 +185,9 @@ async def authorize_payment_gateway(
status = "active",
syncFreq = 60
),
token_notes = {"businessShortCode": inbound_data.auth.businessShortCode},
display_name = inbound_data.auth.businessShortCode,
display_picture = None,
session_token = inbound_headers["X-Session-Token"]
)
api/blueprints/finstitutions/payments/callback.py → api/blueprints/finstitutions/payments/callback_v2.py
+8 -75
View File
@@ -113,7 +113,7 @@ def init(blueprint_setup_state):
# ---------------------------------------------------------------------------------------------------------------------
@pg_callback_bp.route("/callback/safaricom/mpesaexpress", methods = ["POST"])
@pg_callback_bp.route("/callback/safaricom/mpesaexpress", methods = ["GET", "POST"])
@set_api_version(api_version = "1.0.0")
@read_input(sanitize_headers = False, sanitize_data = False)
@log_request_to_mongo(
@@ -144,82 +144,14 @@ async def safaricom_m_pesa_express_callback(
:return: A standard response structure.
"""
# payment_info = await current_app.payment_controller.get_payment_internal(
# mongo_conn = current_app.data_mongo,
# payment_id = ObjectId(payment_id)
# )
# print("PAYMENT INFO:", json.to_string(payment_info.model_dump(), default=str))
# ┏┓ ┓ ┓ ┏┓
# ┣┫┏┫┏┫ ┣ ┓┏┏┓┏┓╋
# ┛┗┗┻┗┻ ┗┛┗┛┗ ┛┗┗
# Map out the documented codes provided by the payment gateway.
# URL: https://developer.safaricom.co.ke/APIs/MpesaExpressSimulate
code_map = {
0: {
"status": "settled",
"message": "Payment successful :)"
}, # ... Success
1037: {
"status": "failed",
"message": "The payment gateway could not reach your customer."
}, # ... DS Timeout. User could not be reached.
1025: {
"status": "failed",
"message": "There was a system error in the payment gateway (1025)."
}, # ... System error while trying to send the push request.
9999: {
"status": "failed",
"message": "There was a system error in the payment gateway (9999)."
}, # ... System error while trying to send the push request.
1032: {
"status": "rejected",
"message": "Your customer declined the payment request."
}, # ... Request Cancelled by the user.
1: {
"status": "failed",
"message": "Your customer has insufficient balance."
}, # ... The user has insufficient balance.
2001: {
"status": "failed",
"message": "The payment gateway says your credentials are invalid."
}, # ... Invalid credentials of the initiator.
1019: {
"status": "failed",
"message": "The transaction expired before your customer processed it."
}, # ... Transaction expired.
1001: {
"status": "failed",
"message": "Your customer is already in the middle of some transaction on the payment gateway."
}, # ... The payer is already making some transaction.
}
# Figure out which of the above codes is relevant to you:
pg_reference_id = inbound_data["Body"]["stkCallback"]["CheckoutRequestID"]
pg_result_code = int(inbound_data["Body"]["stkCallback"]["ResultCode"])
pg_result_desc = inbound_data["Body"]["stkCallback"]["ResultDesc"]
relevant_code = code_map.get(
pg_result_code,
{
"status": "unknown",
"message": f"Unknown code '{pg_result_code}' from the payment gateway. PG: '{pg_result_desc}'"
}
)
# For now, we just insert the event into the record:
event_note_success = await current_app.payment_controller.add_event_by_client_reference_id(
mongo_conn = current_app.data_mongo,
event = PaymentEvent(
paymentStatus = relevant_code["status"],
message = relevant_code["message"],
initByPG = True,
ipAddr = inbound_headers["Remote-IP"],
httpCode = None,
headers = inbound_headers,
payload = inbound_data
),
client_reference_id = pg_reference_id
client_response = await current_app.safaricom_mpesa_express_controller.handle_payment_callback(
mongo_data_conn = current_app.data_mongo,
inbound_data = inbound_data,
inbound_headers = inbound_headers
)
# ┳┓
@@ -229,8 +161,9 @@ async def safaricom_m_pesa_express_callback(
# Done here:
return ResponseModel(
status_code = StatusCodes.OK if event_note_success else StatusCodes.FAILED,
http_code = HttpCodes.SUCCESS if event_note_success else HttpCodes.INTERNAL_SERVER_ERROR
status_code = StatusCodes.OK if client_response.success else StatusCodes.FAILED,
http_code = HttpCodes.SUCCESS if client_response.success else HttpCodes.INTERNAL_SERVER_ERROR,
message = client_response.message
)
api/blueprints/finstitutions/payments/get.py → api/blueprints/finstitutions/payments/get_v2.py
+4 -4
View File
@@ -6,7 +6,7 @@
DATE:
Wednesday, 18th Dec., 2024
Saturday, 28th Dec., 2024
OBJECTIVE:
@@ -177,8 +177,8 @@ async def get_one_payment_record(
# ┻ ┗ ┗┗┛┗ ┛┗┗ ┗┗┛┛ ┗┻
# Get the payment record:
record = await current_app.payment_controller.get_payment(
mongo_conn = current_app.data_mongo,
record = await current_app.payments_controller.get_payment(
mongo_data_conn = current_app.data_mongo,
payment_id = inbound_data.paymentId
)
@@ -189,7 +189,7 @@ async def get_one_payment_record(
# We check if the token that was used to fetch the mail is owned by this user:
if not await token_check.is_authorized(
mongo_conn = current_app.data_mongo,
mongo_data_conn = current_app.data_mongo,
user_info = CoreUserInfoModel(**kwargs["session_info"]),
token_ids = [record.tokenId]
): return ResponseModel(
api/blueprints/finstitutions/payments/list.py → api/blueprints/finstitutions/payments/list_v2.py
+9 -12
View File
@@ -6,7 +6,7 @@
DATE:
Tuesday, 17th Dec., 2024
Saturday, 28th Dec., 2024
OBJECTIVE:
@@ -161,30 +161,27 @@ async def payment_list(
message = "invalid session"
)
# ┏┓ ┓• ┳┳┓ •┓
# ┣ ┏┓┃┓┏╋ ┃┃┃┏┓┓┃┏
# ┗┛┛┗┗┗┛┗ ┛ ┗┗┻┗┗┛
# ┏┓ ┓• ┏┓
# ┣ ┏┓┃┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏
# ┗┛┛┗┗┗┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛
# ┛
# Get the token ids from the token keys:
auth_tokens = await current_app.payment_controller.get_tokens_from_keys(
mongo_conn = current_app.data_mongo,
auth_tokens = await current_app.payments_controller.get_tokens_from_keys(
mongo_data_conn = current_app.data_mongo,
token_keys = inbound_data.tokenKeys
)
token_ids = [t.authTokenId for t in auth_tokens]
print("TOKEN IDS:", json.to_string(token_ids, default = str))
# Build the additional filter:
additional_filter = {}
if inbound_data.tags: additional_filter["tags"] = {"$in": inbound_data.tags}
if inbound_data.paymentStatus: additional_filter["lastPaymentStatus"] = {"$in": inbound_data.paymentStatus}
additional_filter = additional_filter or None
print("ADDFIL:", json.to_string(additional_filter))
# Get the mails:
payment_records = await current_app.payment_controller.list_payments(
mongo_conn = current_app.data_mongo,
payment_records = await current_app.payments_controller.list_payments(
mongo_data_conn = current_app.data_mongo,
token_ids = token_ids,
limit = inbound_data.count,
skip = inbound_data.fromCount,
api/blueprints/finstitutions/payments/request.py → api/blueprints/finstitutions/payments/request_v2.py
+23 -16
View File
@@ -6,7 +6,7 @@
DATE:
Monday, 16th Dec., 2024
Saturday, 28th Dec., 2024
OBJECTIVE:
@@ -161,29 +161,36 @@ async def request_payment(
http_code = HttpCodes.UNAUTHORIZED
)
# ┳┓ ┏┓
# ┣┫┏┓┏┓┓┏┏┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋
# ┛┗┗ ┗┫┗┻┗ ┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗
# ┗ ┛
# Get the token from the token key:
auth_token = await current_app.payment_controller.get_token_from_key(
mongo_conn = current_app.data_mongo,
auth_token = await current_app.payments_controller.get_token_from_key(
mongo_data_conn = current_app.data_mongo,
token_key = inbound_data.tokenKey
)
if auth_token is None: return ResponseModel(
status_code = StatusCodes.FAILED,
http_code = HttpCodes.UNAUTHORIZED,
message = f"no such token key"
message = f"No such token key."
)
# Make the request with the fetched token:
response = await current_app.payment_controller.request_payment(
mongo_conn = current_app.data_mongo,
http_client = current_app.http_client,
auth_token = auth_token,
user_info = CoreUserInfoModel(**kwargs["session_info"]),
payment_request = inbound_data
# ┳┓ ┏┓
# ┣┫┏┓┏┓┓┏┏┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋
# ┛┗┗ ┗┫┗┻┗ ┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗
# ┗ ┛
# For Safaricom's M-Pesa Express client:
if auth_token.client == "safaricomMPesaExpress":
response = await current_app.safaricom_mpesa_express_controller.request_payment(
mongo_data_conn = current_app.data_mongo,
auth_token = auth_token,
user_info = CoreUserInfoModel(**kwargs["session_info"]),
payment_request = inbound_data
)
# Invalid client:
else: return ResponseModel(
status_code = StatusCodes.FAILED,
http_code = HttpCodes.BAD_REQUEST,
message = f"Invalid/unimplemented client '{auth_token.client}'"
)
# ┳┓
api/blueprints/finstitutions/payments/tags.py → api/blueprints/finstitutions/payments/tags_v2.py
+12 -7
View File
@@ -6,7 +6,7 @@
DATE:
Wednesday, 18th Dec., 2024
Saturday, 28th Dec., 2024
OBJECTIVE:
@@ -176,10 +176,15 @@ async def update_payment_record_tags(
# ┻ ┗ ┗┗┛┗ ┛┗┗ ┗┗┛┛ ┗┻
# Get the payment record:
record = await current_app.payment_controller.get_payment(
mongo_conn = current_app.data_mongo,
record = await current_app.payments_controller.get_payment(
mongo_data_conn = current_app.data_mongo,
payment_id = inbound_data.paymentId
)
if not record: return ResponseModel(
status_code = StatusCodes.FAILED,
http_code = HttpCodes.NOT_FOUND,
message = "No matching payment record found."
)
# ┏┓ ┓ • ┏┓┓ ┓
# ┃┃┓┏┏┏┓┏┓┏┓┏┣┓┓┏┓ ┃ ┣┓┏┓┏┃┏
@@ -188,13 +193,13 @@ async def update_payment_record_tags(
# We check if the token that was used to fetch the mail is owned by this user:
if not await token_check.is_authorized(
mongo_conn = current_app.data_mongo,
mongo_data_conn = current_app.data_mongo,
user_info = CoreUserInfoModel(**kwargs["session_info"]),
token_ids = [record.tokenId]
): return ResponseModel(
status_code = StatusCodes.FAILED,
http_code = HttpCodes.UNAUTHORIZED,
message = "The record does not belong to this user."
message = "The payment record does not belong to this user."
)
# ┳┳ ┓ ┳┓ ┓
@@ -203,8 +208,8 @@ async def update_payment_record_tags(
# ┛
# Update the record:
success = await current_app.payment_controller.update_tags(
mongo_conn = current_app.data_mongo,
success = await current_app.payments_controller.update_tags(
mongo_data_conn = current_app.data_mongo,
payment_id = inbound_data.paymentId,
unset_tags = inbound_data.unsetTags,
set_tags = inbound_data.setTags
api/helpers/user/token_check.py
+6 -6
View File
@@ -88,14 +88,14 @@ from models.core.user import CoreUserInfoModel
async def is_authorized(
mongo_conn: AsyncMongo,
mongo_data_conn: AsyncMongo,
user_info: CoreUserInfoModel | dict,
token_ids: ObjectId | str | List[ObjectId | str]
) -> bool:
"""
To verify if the token id is owned by the user (who will be initially identified from his session token).
:param mongo_conn: The instance of the MongoDB connector to use to perform this action.
:param mongo_data_conn: The instance of the MongoDB connector to use to perform this action.
:param user_info: The user trying to access some service.
:param token_ids: One or more token ids that the user is claiming to own.
:return: True if the user is authorized to use this token, else False.
@@ -113,7 +113,7 @@ async def is_authorized(
# Fetch the auth tokens from the database:
# Retrieve the document of the token id:
auth_tokens = await current_app.core_auth_token_controller.get_tokens_from_ids(
mongo_data_conn = mongo_conn,
mongo_data_conn = mongo_data_conn,
token_ids = [ObjectId(t) for t in token_ids]
)
@@ -131,21 +131,21 @@ async def is_authorized(
async def is_not_authorized(
mongo_conn: AsyncMongo,
mongo_data_conn: AsyncMongo,
user_info: CoreUserInfoModel,
token_ids: ObjectId | str | List[ObjectId | str]
) -> bool:
"""
Just a wrapper around the above function to improve readability.
:param mongo_conn: The instance of the MongoDB connector to use to perform this action.
:param mongo_data_conn: The instance of the MongoDB connector to use to perform this action.
:param user_info: The user trying to access some service.
:param token_ids: One or more token ids that the user is claiming to own.
:return: True if the user is authorized to use this token, else False.
"""
authorized = await is_authorized(
mongo_conn = mongo_conn,
mongo_data_conn = mongo_data_conn,
user_info = user_info,
token_ids = token_ids
)
api/main.py
+33 -16
View File
@@ -71,7 +71,7 @@ from controllers.core.payment import CorePaymentController
# API Controller Models:
from controllers.api.mail import MailController
# from controllers.api.sms import SMSController
from controllers.api.payment import PaymentController
# from controllers.api.payment import PaymentController
# Controllers V2:
from controllers_v2.core.auth_token import CoreAuthTokenController
@@ -82,6 +82,9 @@ from controllers_v2.message.sms.savvy_bulk_sms_kenya import SavvyBulkSMSKenyaCon
# ---
from controllers_v2.finstitutions.trading.all_trading import AllTradingController
from controllers_v2.finstitutions.trading.zerodha_kite import ZerodhaKiteTradingController
# ---
from controllers_v2.finstitutions.payments.all_payments import AllPaymentsController
from controllers_v2.finstitutions.payments.safaricom_mpesa_express import SafaricomMPesaExpressPaymentsController
# To make REST API calls:
import httpx
@@ -112,12 +115,12 @@ from api.blueprints.sms.tags import sms_update_tags_bp
from api.blueprints.software.auth import sw_auth_bp
# Finstitutions / Payment Blueprints:
from api.blueprints.finstitutions.payments.auth import pg_auth_bp
from api.blueprints.finstitutions.payments.request import pg_request_bp
from api.blueprints.finstitutions.payments.callback import pg_callback_bp
from api.blueprints.finstitutions.payments.list import pg_list_bp
from api.blueprints.finstitutions.payments.get import pg_get_bp
from api.blueprints.finstitutions.payments.tags import pg_tags_update_bp
from api.blueprints.finstitutions.payments.auth_v2 import pg_auth_bp
from api.blueprints.finstitutions.payments.request_v2 import pg_request_bp
from api.blueprints.finstitutions.payments.callback_v2 import pg_callback_bp
from api.blueprints.finstitutions.payments.list_v2 import pg_list_bp
from api.blueprints.finstitutions.payments.get_v2 import pg_get_bp
from api.blueprints.finstitutions.payments.tags_v2 import pg_tags_update_bp
# Finstitutions / Trading Blueprints:
from api.blueprints.finstitutions.trading.oauth.request import trading_oauth_request_bp
@@ -401,14 +404,14 @@ async def app_startup(**kwargs):
debug_prefix = "Message (CM) | ",
debug_only_errors = True
)
current_app.core_payment_controller = CorePaymentController(
cache = current_app.module_cache,
alert_url = current_app.script_data["alerts"]["url"],
http_client = current_app.http_client,
debug = enable_debugging,
debug_prefix = "Pymnt. (CM) | ",
debug_only_errors = True
)
# current_app.core_payment_controller = CorePaymentController(
# cache = current_app.module_cache,
# alert_url = current_app.script_data["alerts"]["url"],
# http_client = current_app.http_client,
# debug = enable_debugging,
# debug_prefix = "Pymnt. (CM) | ",
# debug_only_errors = True
# )
# ┏┓┏┓┳ ┏┓ ┓┓
# ┣┫┃┃┃ ┃ ┏┓┏┓╋┏┓┏┓┃┃┏┓┏┓┏
@@ -416,7 +419,7 @@ async def app_startup(**kwargs):
current_app.mail_controller = MailController()
# current_app.sms_controller = SMSController()
current_app.payment_controller = PaymentController()
# current_app.payment_controller = PaymentController()
# ┏┓ ┓┓ ┓┏┏┓
# ┃ ┏┓┏┓╋┏┓┏┓┃┃┏┓┏┓┏ ┃┃┏┛
@@ -464,6 +467,20 @@ async def app_startup(**kwargs):
debug = enable_debugging
)
# Finstitutions / Payments Controllers:
current_app.payments_controller = AllPaymentsController(
cache = current_app.module_cache,
http_client = current_app.http_client,
alert_url = current_app.script_data["alerts"]["url"],
debug = enable_debugging
)
current_app.safaricom_mpesa_express_controller = SafaricomMPesaExpressPaymentsController(
cache = current_app.module_cache,
http_client = current_app.http_client,
alert_url = current_app.script_data["alerts"]["url"],
debug = enable_debugging
)
# ┏┓ ┓ ┏┓┓•
# ┃ ┏┓┏┓┏┓┏┓┏╋┏┓┏┓┏ ┏┓┏┓┏┫ ┃ ┃┓┏┓┏┓╋┏
# ┗┛┗┛┛┗┛┗┗ ┗┗┗┛┛ ┛ ┗┻┛┗┗┻ ┗┛┗┗┗ ┛┗┗┛
controllers_v2/finstitutions/payments/__init__.py
View File
controllers_v2/finstitutions/payments/all_payments.py
+186
View File
@@ -0,0 +1,186 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Saturday, 28th Dec., 2024
OBJECTIVE:
To handle all common actions related to payments from one place. This includes cases where you don't yet know
the third-party client or it doesn't matter who the third-party client is. For example, take those cases when
you just need to fetch one record about a payment.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# My async utils:
from utils_v2.database.async_mysql_v2 import AsyncMySQL
from utils_v2.database.async_mongo_v2 import AsyncMongo
from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache
# Controllers:
from controllers_v2.finstitutions.payments.base import PaymentsController
# Models:
from models.core.user import CoreUserInfoModel
from models.core.auth_token import CoreAuthTokenModel
from models.api.finstitutions.payments.request import PGPaymentRequestData, PaymentRequestOneResult
# To work with datatypes:
from typing import List, Any
# To make HTTP requests:
import httpx
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class AllPaymentsController(PaymentsController):
# ┏┓
# ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓
# ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛
def __init__(
self,
cache: AsyncRedisCache = None,
http_client: httpx.AsyncClient = None,
alert_url: str = None,
debug: bool = True,
debug_prefix: str = "All Payments (C) | ",
debug_only_errors: bool = True
):
"""
This is the foundational controller for all trading services. Use this for any smaller common tasks where you
may not know the exact client beforehand.
:param cache: The object to use for caching results from database calls.
:param http_client: The HTTP client
:param debug: Whether, or not, you would like to print debugging messages:
:param debug_prefix: The prefix to print with the debugging messages.
:param debug_only_errors: Whether you would like to print only error messages or all messages.
:return: None.
"""
# Invoke the parent's constructor:
super().__init__(
cache = cache,
alert_url = alert_url,
http_client = http_client,
base_filter = None,
debug = debug,
debug_prefix = debug_prefix,
debug_only_errors = debug_only_errors
)
# ┳┓ ┏┓
# ┣┫┏┓┏┓┓┏┏┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏
# ┛┗┗ ┗┫┗┻┗ ┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛
# ┗ ┛
async def request_payment(
self,
mongo_data_conn: AsyncMongo,
auth_token: CoreAuthTokenModel,
user_info: CoreUserInfoModel,
payment_request: PGPaymentRequestData,
) -> PaymentRequestOneResult:
"""
To request payment from someone through a payment gateway.
:param mongo_data_conn: The database connection to use to perform this activity.
:param auth_token: The token that has to be used to fetch the data.
:param user_info: The info. of your user, so that you can identify who requested the payment.
:param payment_request: The data that came in with the APi call.
:return: The structured response form the payment gateway.
"""
raise NotImplementedError
async def handle_payment_callback(
self,
mongo_data_conn: AsyncMongo,
inbound_data: dict
):
"""
Whenever the payment gateway sends an update about a requested payment, we use this method to update our records
as per the specification of the third-party payment gateway.
:param mongo_data_conn: The database connection to use to perform this activity.
:param inbound_data: The data sent by the payment gateway in their update.
:return: ??
"""
raise NotImplementedError
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
controllers_v2/finstitutions/payments/base.py
+515
View File
@@ -0,0 +1,515 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Saturday, 28th Dec., 2024
OBJECTIVE:
To handle all payments-related behaviour from one place.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# My async utils:
from utils_v2.database.async_mysql_v2 import AsyncMySQL
from utils_v2.database.async_mongo_v2 import AsyncMongo
from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache
# Controllers:
from controllers_v2.core.auth_token import CoreAuthTokenController
# Models:
from models.core.user import CoreUserInfoModel
from models.core.auth_token import CoreAuthTokenModel
from models.core.payment import CorePaymentModel, PaymentEvent, CustomerDetails
from models.api.finstitutions.payments.request import PGPaymentRequestData, PaymentRequestOneResult
# To work with MongoDB:
from bson import ObjectId
# To work with datatypes:
from typing import List, Any
# To make HTTP requests:
import httpx
# To make abstract classes:
from abc import ABC, abstractmethod
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class PaymentsController(CoreAuthTokenController, ABC):
# ┏┓┓ ┓┏
# ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏
# ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛
# For MongoDB:
PAYMENTS_COLLECTION = "_payments"
# ┏┓
# ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓
# ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛
def __init__(
self,
cache: AsyncRedisCache = None,
http_client: httpx.AsyncClient = None,
alert_url: str = None,
base_filter: dict = None,
debug: bool = True,
debug_prefix: str = "Payments (C) | ",
debug_only_errors: bool = True
):
"""
This is the foundational controller for all trading/stockbroking services. This is built on top of the
authorization model, and, in turn, the individual stockbroking clients should be built on top of this.
:param cache: The object to use for caching results from database calls.
:param http_client: The HTTP client
:param base_filter: The basic filter that will be applied to all fetching/updating queries. WARNING: THE BASE
FILTER WILL ALWAYS BE APPLIED AUTOMATICALLY. SET THIS UP WISELY.
:param debug: Whether, or not, you would like to print debugging messages:
:param debug_prefix: The prefix to print with the debugging messages.
:param debug_only_errors: Whether you would like to print only error messages or all messages.
:return: None.
"""
# Declare the service type:
this_service_type = "paymentGateway"
# Prepare base filter:
this_filter = {}
for k, v in (base_filter or {}).items(): this_filter[k] = v
this_filter["serviceType"] = this_service_type
# Invoke the parent's constructor:
CoreAuthTokenController.__init__(
self,
cache = cache,
alert_url = alert_url,
http_client = http_client,
base_filter = this_filter,
debug = debug,
debug_prefix = debug_prefix,
debug_only_errors = debug_only_errors
)
# Init a variable in a parent:
self._service_type = this_service_type
# ┓ • ┏┓
# ┃ ┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏
# ┗┛┗┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛
# ┛
# These are simply for retrieving payment records.
# You need to already have them saved to the database.
async def count_payments(
self,
mongo_data_conn: AsyncMongo,
token_ids: List[ObjectId | str],
additional_filter: dict = None
) -> int:
"""
Just counts the no. of payment records that match a given set of conditions.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param token_ids: The token ids of the accounts from which these payment details must be fetched.
:param additional_filter: Any addition filters to use.
:return: The no. of payment records that match the given conditions.
"""
# Prepare the filter:
if not isinstance(token_ids, list): token_ids = [token_ids]
token_ids = [ObjectId(t) for t in token_ids]
filter_json = {"tokenId": {"$in": token_ids}}
if additional_filter:
for k, v in additional_filter.items():
filter_json[k] = v
# Get the count of the documents that match the criteria:
count = await mongo_data_conn.count(
collection = self.PAYMENTS_COLLECTION,
filter = filter_json,
raise_exception = True
)
# Done here:
return count
async def list_payments(
self,
mongo_data_conn: AsyncMongo,
token_ids: List[ObjectId | str],
limit: int = 100,
skip: int = 0,
additional_filter: dict = None
) -> List[CorePaymentModel] | None:
"""
Fetches many payment details in one call, but just their previews.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param token_ids: The token ids of the accounts from which these messages must be fetched.
:param limit: The max. no. of payment details to retrieve in this call.
:param skip: The no. of initial payment details to skip. Useful for pagination.
:param additional_filter: Any addition filters to use.
:return: The list of payments (as the payments model). This list can be empty.
"""
# Prepare the filter:
if not isinstance(token_ids, list): token_ids = [token_ids]
token_ids = [ObjectId(t) for t in token_ids]
filter_json = {"tokenId": {"$in": token_ids}}
if additional_filter:
for k, v in additional_filter.items():
filter_json[k] = v
# We fetch the messages that are identified by a specific token id,
# with the specified fetching limits, while enforcing the sorting condition:
records = await mongo_data_conn.find_many(
collection = self.PAYMENTS_COLLECTION,
filter = filter_json,
projection = {"events": False},
limit = limit,
skip = skip,
sort = {"ts": -1},
raise_exception = True
)
# Convert the fetched records to instances of the data model and return:
for record in records: record["events"] = []
return [CorePaymentModel(**record) for record in records]
async def get_payment(
self,
mongo_data_conn: AsyncMongo,
payment_id: ObjectId | str
) -> CorePaymentModel | None:
"""
Gets one payment detail if you know its payment id.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param payment_id: The id of the payment detail that needs to be read.
:return: The contents of that one payment detail in a structured format.
"""
# We fetch the whole payload of that one message:
record = await mongo_data_conn.find_one(
collection = self.PAYMENTS_COLLECTION,
filter = {"_id": ObjectId(payment_id)},
raise_exception = True
)
# If no such message was found:
if record is None: return None
# If a record was found,
# we return it as our data model:
return CorePaymentModel(**record)
# ┳┳ ┓ ┏┓
# ┃┃┏┓┏┫┏┓╋┏┓ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏
# ┗┛┣┛┗┻┗┻┗┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛
# ┛ ┛
# We don't support updating payments themselves,
# but we will allow updating fields like tags, adding events, etc.
async def add_event_by_payment_id(
self,
mongo_data_conn: AsyncMongo,
payment_id: ObjectId | str,
event: PaymentEvent,
client_reference_id: str = None
) -> bool:
"""
Add an event to an existing record of a payment detail.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param payment_id: The id of the payment detail that needs to be read.
:param event: The event that occurred. This will typically be generated by the third-party client.
:param client_reference_id: The way the client identifies this payment. You need to pass this only on the first
event. Typically, when you initiate the payment request.
:return: True if successfully noted, else False.
"""
# Prepare the update document:
update_json = {
"$push": {
"events": event.model_dump()
},
"$set": {
"lastEventTs": event.eventTs,
"lastEventMessage": event.message,
"lastPaymentStatus": event.paymentStatus,
}
}
if client_reference_id: update_json["$set"]["clientPaymentReferenceId"] = client_reference_id
# Try to update the existing record:
return await mongo_data_conn.update_one(
collection = self.PAYMENTS_COLLECTION,
filter = {"_id": ObjectId(payment_id)},
update = update_json,
upsert = False,
raise_exception = True
)
async def add_event_by_client_reference_id(
self,
mongo_data_conn: AsyncMongo,
client_reference_id: str,
event: PaymentEvent,
) -> bool:
"""
Add an event to an existing record of a payment detail.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param event: The event that occurred. This will typically be generated by the third-party client.
:param client_reference_id: The way the client identifies this payment. You need to pass this only on the first
event. Typically, when you initiate the payment request.
:return: True if successfully noted, else False.
"""
# Prepare the update document:
update_json = {
"$push": {
"events": event.model_dump()
},
"$set": {
"lastEventTs": event.eventTs,
"lastEventMessage": event.message,
"lastPaymentStatus": event.paymentStatus
}
}
if client_reference_id: update_json["$set"]["clientPaymentReferenceId"] = client_reference_id
# Try to update the existing record:
return await mongo_data_conn.update_one(
collection = self.PAYMENTS_COLLECTION,
filter = {"clientPaymentReferenceId": client_reference_id},
update = update_json,
upsert = False,
raise_exception = True
)
async def update_tags(
self,
mongo_data_conn: AsyncMongo,
payment_id: ObjectId | str,
unset_tags: List[str] = None,
set_tags: List[str] = None
) -> bool:
"""
Updates the tags on one payment. The tags to remove are processed first, the ones to add are processed later.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param payment_id: The id of the payment detail that needs to be read.
:param unset_tags: The tags to remove from the payment record.
:param set_tags: The tags to add to the payment record.
:return: True if the update was successful, else False.
"""
# Update the tags:
return await mongo_data_conn.update_one(
collection = self.PAYMENTS_COLLECTION,
filter = {"_id": ObjectId(payment_id)},
update = [{
"$set": {
"tags": {
"$let": {
"vars": {
"removed_tags": {
"$setDifference": [
"$tags",
unset_tags
]
}
},
"in": {
"$setUnion": [
"$$removed_tags",
set_tags
]
}
}
}
}
}],
raise_exception = True
)
# ┳┓ ┓ ┏┓
# ┃┃┏┓┃┏┓╋┏┓ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏
# ┻┛┗ ┗┗ ┗┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛
# ┛
# No support whatsoever for deleting payment records!
# ┳┓ ┏┓
# ┣┫┏┓┏┓┓┏┏┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏
# ┛┗┗ ┗┫┗┻┗ ┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛
# ┗ ┛
async def init_payment(
self,
mongo_data_conn: AsyncMongo,
auth_token: CoreAuthTokenModel,
user_info: CoreUserInfoModel,
payment_request: PGPaymentRequestData,
tags: List[Any]
) -> ObjectId | None:
"""
Do this before you hit the third-party client's service when requesting payments. This creates a validated
payment record in the database which can then be used as reference for successive updates.
:param mongo_data_conn: The database connection to use to perform this activity.
:param auth_token: The token that has to be used to fetch the data.
:param user_info: The user of your platform, NOT THE PAYING PARTY.
:param payment_request: The payment request that came in through the API call.
:param tags: And initial tags to apply to this payment's records that you know you will need for filtering.
:return: The id of the MongoDB document that will hold the full record of this payment.
"""
# Model the payment's request. This ensures we're validating the inputs.
payment_model = CorePaymentModel(
user = user_info,
customer = CustomerDetails(
name = payment_request.customerName,
contactNo = payment_request.customerNo,
payerNo = payment_request.payerNo,
email = payment_request.email
),
lastEventMessage = "Payment Request Queued",
lastPaymentStatus = "queued",
tokenId = auth_token.authTokenId,
amount = payment_request.amount,
currencyCode = payment_request.currencyCode,
metadata = payment_request.metadata.model_dump(),
tags = list(set(payment_request.tags + (tags or []))),
serviceType = "paymentGateway",
client = auth_token.client,
clientPaymentReferenceId = None,
events = []
)
# Simply insert the document and return the id:
return await mongo_data_conn.insert_one(
collection = self.PAYMENTS_COLLECTION,
document = payment_model.model_dump(),
raise_exception = True
)
@abstractmethod
async def request_payment(
self,
mongo_data_conn: AsyncMongo,
auth_token: CoreAuthTokenModel,
user_info: CoreUserInfoModel,
payment_request: PGPaymentRequestData,
) -> PaymentRequestOneResult:
"""
To request payment from someone through a payment gateway.
:param mongo_data_conn: The database connection to use to perform this activity.
:param auth_token: The token that has to be used to fetch the data.
:param user_info: The info. of your user, so that you can identify who requested the payment.
:param payment_request: The data that came in with the APi call.
:return: The structured response form the payment gateway.
"""
pass
@abstractmethod
async def handle_payment_callback(
self,
mongo_data_conn: AsyncMongo,
inbound_data: dict,
inbound_headers: dict
) -> None:
"""
Whenever the payment gateway sends an update about a requested payment, we use this method to update our records
as per the specification of the third-party payment gateway.
:param mongo_data_conn: The database connection to use to perform this activity.
:param inbound_data: The data sent by the payment gateway in their update.
:param inbound_headers: The headers sent by the payment gateway in their update.
:return: ??
"""
pass
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
controllers_v2/finstitutions/payments/safaricom_mpesa_express.py
+356
View File
@@ -0,0 +1,356 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Saturday, 28th Dec., 2024
OBJECTIVE:
To handle all interactions with Safaricom's M-Pesa Express payment gateway from one place.
REFERENCES:
01. Official Documentation: https://developer.safaricom.co.ke/APIs/MpesaExpressSimulate
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# My async utils:
from utils_v2.string import json
from utils_v2.string import regex
from utils_v2.database.async_mysql_v2 import AsyncMySQL
from utils_v2.database.async_mongo_v2 import AsyncMongo
from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache
# Controllers:
from controllers_v2.finstitutions.payments.base import PaymentsController
# Models:
from models.core.user import CoreUserInfoModel
from models.core.auth_token import CoreAuthTokenModel
from models.core.payment import CorePaymentModel, PaymentEvent, CustomerDetails
from models.api.finstitutions.payments.request import (
PGPaymentRequestData,
PaymentRequestOneResult,
PaymentCallbackResult
)
# Payment Client:
from utils_v2.payments.safaricom.models.auth import MPesaExpressAuthorization
from utils_v2.payments.safaricom.controllers.m_pesa_express import SafaricomMPesaExpress
# To work with datatypes:
from typing import List, Any
# To make HTTP requests:
import httpx
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class SafaricomMPesaExpressPaymentsController(PaymentsController):
# ┏┓
# ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓
# ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛
def __init__(
self,
cache: AsyncRedisCache = None,
http_client: httpx.AsyncClient = None,
alert_url: str = None,
debug: bool = True,
debug_prefix: str = "Sfrcm. M-Pesa Exp. (C) | ",
debug_only_errors: bool = True
):
"""
This is the foundational controller for all trading services. Use this for any smaller common tasks where you
may not know the exact client beforehand.
:param cache: The object to use for caching results from database calls.
:param http_client: The HTTP client
:param debug: Whether, or not, you would like to print debugging messages:
:param debug_prefix: The prefix to print with the debugging messages.
:param debug_only_errors: Whether you would like to print only error messages or all messages.
:return: None.
"""
# Invoke the parent's constructor:
super().__init__(
cache = cache,
alert_url = alert_url,
http_client = http_client,
base_filter = None,
debug = debug,
debug_prefix = debug_prefix,
debug_only_errors = debug_only_errors
)
# ┓┏ ┓
# ┣┫┏┓┃┏┓┏┓┏┓┏
# ┛┗┗ ┗┣┛┗ ┛ ┛
# ┛
@staticmethod
def clean_phone_no(value: str) -> str:
"""
To clean-up input Kenyan phone nos.
:param value: The phone no. to clean, provided as a string.
:return: The cleaned phone no.
"""
return regex.replace(
text = value,
pattern = r"[^0-9]",
substitute_text = ""
)
# ┳┓ ┏┓
# ┣┫┏┓┏┓┓┏┏┓┏╋ ┃┃┏┓┓┏┏┳┓┏┓┏┓╋┏
# ┛┗┗ ┗┫┗┻┗ ┛┗ ┣┛┗┻┗┫┛┗┗┗ ┛┗┗┛
# ┗ ┛
async def request_payment(
self,
mongo_data_conn: AsyncMongo,
auth_token: CoreAuthTokenModel,
user_info: CoreUserInfoModel,
payment_request: PGPaymentRequestData
) -> PaymentRequestOneResult:
"""
To request payment from someone through a payment gateway.
:param mongo_data_conn: The database connection to use to perform this activity.
:param auth_token: The token that has to be used to fetch the data.
:param user_info: The info. of your user, so that you can identify who requested the payment.
:param payment_request: The data that came in with the APi call.
:return: The structured response form the payment gateway.
"""
# Start by assuming failure:
result = PaymentRequestOneResult()
# Initialize the payment request by creating a placeholder record in the database:
payment_id = await self.init_payment(
mongo_data_conn = mongo_data_conn,
auth_token = auth_token,
user_info = user_info,
payment_request = payment_request,
tags = ["Payment", "Safaricom", "M-Pesa Express", "Kenya"]
)
# Initialize the third-party client:
client = SafaricomMPesaExpress(
auth = MPesaExpressAuthorization(
consumerKey = auth_token.auth["consumerKey"],
consumerSecret = auth_token.auth["consumerSecret"],
businessShortCode = auth_token.auth["businessShortCode"],
appPasskey = auth_token.auth["appPasskey"]
),
http_client = self._http_client
)
# Make the payment request:
client_response = await client.request_payment(
amount = payment_request.amount,
party_a = self.clean_phone_no(
payment_request.customerNo
) if isinstance(payment_request.customerNo, str) else payment_request.customerNo,
type = "CustomerPayBillOnline",
reference = str(payment_id),
description = payment_request.description,
callback_url = f"https://api.thecaoffice.com/finstitutions/payments/callback/safaricom/mpesaexpress",
payer_no = self.clean_phone_no(
payment_request.payerNo
) if isinstance(payment_request.payerNo, str) else payment_request.payerNo,
party_b = auth_token.auth["businessShortCode"]
)
# Add this event to the payment's document:
event_note_success = await self.add_event_by_payment_id(
mongo_data_conn = mongo_data_conn,
payment_id = payment_id,
event = PaymentEvent(
paymentStatus = "initiated" if client_response.success else "initFailed",
message = f"PG: {client_response.message}",
initByPG = False,
httpCode = client_response.httpCode,
headers = await client_response.get_headers(),
payload = await client_response.get_json()
),
client_reference_id = client_response.referenceId
)
# Done here:
result.success = client_response.success and event_note_success
result.message = " ".join([
"Payment requested successfully." if client_response.success
else f"Payment request FAILED (PG: '{client_response.message}').",
" " if event_note_success
else "Event noting FAILED.",
]).strip()
return result
async def handle_payment_callback(
self,
mongo_data_conn: AsyncMongo,
inbound_data: dict,
inbound_headers: dict
) -> PaymentCallbackResult:
"""
Whenever the payment gateway sends an update about a requested payment, we use this method to update our records
as per the specification of the third-party payment gateway.
:param mongo_data_conn: The database connection to use to perform this activity.
:param inbound_data: The data sent by the payment gateway in their update.
:param inbound_headers: The headers sent by the payment gateway in their update.
:return: A structured response about the process of updating the payment event.
"""
# Start by assuming failure:
result = PaymentCallbackResult()
# Map out the documented codes provided by the payment gateway.
# URL: https://developer.safaricom.co.ke/APIs/MpesaExpressSimulate
code_map = {
0: {
"status": "settled",
"message": "Payment successful :)"
}, # ... Success
1037: {
"status": "failed",
"message": "The payment gateway could not reach your customer."
}, # ... DS Timeout. User could not be reached.
1025: {
"status": "failed",
"message": "There was a system error in the payment gateway (1025)."
}, # ... System error while trying to send the push request.
9999: {
"status": "failed",
"message": "There was a system error in the payment gateway (9999)."
}, # ... System error while trying to send the push request.
1032: {
"status": "rejected",
"message": "Your customer declined the payment request."
}, # ... Request Cancelled by the user.
1: {
"status": "failed",
"message": "Your customer has insufficient balance."
}, # ... The user has insufficient balance.
2001: {
"status": "failed",
"message": "The payment gateway says your credentials are invalid."
}, # ... Invalid credentials of the initiator.
1019: {
"status": "failed",
"message": "The transaction expired before your customer processed it."
}, # ... Transaction expired.
1001: {
"status": "failed",
"message": "Your customer is already in the middle of some transaction on the payment gateway."
}, # ... The payer is already making some transaction.
}
# Figure out which of the above codes is relevant to you:
pg_reference_id = inbound_data["Body"]["stkCallback"]["CheckoutRequestID"]
pg_result_code = int(inbound_data["Body"]["stkCallback"]["ResultCode"])
pg_result_desc = inbound_data["Body"]["stkCallback"]["ResultDesc"]
relevant_code = code_map.get(
pg_result_code,
{
"status": "unknown",
"message": f"Unknown code '{pg_result_code}' from the payment gateway. PG: '{pg_result_desc}'"
}
)
# For now, we just insert the event into the record:
event_note_success = await self.add_event_by_client_reference_id(
mongo_data_conn = mongo_data_conn,
event = PaymentEvent(
paymentStatus = relevant_code["status"],
message = relevant_code["message"],
initByPG = True,
ipAddr = inbound_headers["Remote-IP"],
httpCode = None,
headers = inbound_headers,
payload = inbound_data
),
client_reference_id = pg_reference_id
)
# Done here:
if event_note_success:
result.success = True
result.message = f"Payment event noted."
else:
result.success = False
result.message = f"Payment event NOT noted."
return result
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
models/api/finstitutions/payments/request.py
+25 -1
View File
@@ -230,7 +230,31 @@ class PGPaymentRequestData(BaseModel):
class PaymentRequestOneResult(BaseModel):
success: bool = Field(
description = "whether, or not, the sms was successfully sent",
description = "whether, or not, the payment was successfully requested",
default = False
)
message: str | None = Field(
description = "a brief message to summarize the result of the process",
default = None
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ---------------------------------------------------------------------------------------------------------------------
class PaymentCallbackResult(BaseModel):
success: bool = Field(
description = "whether, or not, the payment event was noted",
default = False
)
models/core/payment.py
+2 -1
View File
@@ -216,7 +216,8 @@ class CorePaymentModel(BaseModel):
description = "the id of the document in mongodb that holds this information",
frozen = True,
default = None,
alias = "_id"
alias = "_id",
exclude = True
)
user: CoreUserInfoModel = Field(
models/finstitutions/trading/ticks.py
+6
View File
@@ -155,6 +155,10 @@ class TradingTick(BaseModel):
description = "the symbol of the instrument"
)
name: str = Field(
description = "the name of the co./underlying"
)
exchange: Literal["NSE", "NFO", "BSE", "BFO", "MCX", "CDS", "BCD"] = Field(
description = "the exchange on which this instrument is traded",
frozen = True
@@ -357,6 +361,7 @@ class TradingTick(BaseModel):
"segment": self.segment,
"type": self.type,
"symbol": self.symbol,
"name": self.name,
"expiry": date_time.to_timezone(
self.expiryTs,
timezone = self.expiryTz
@@ -398,6 +403,7 @@ class TradingTick(BaseModel):
modelled_ticks.append(
TradingTick(
symbol = tick_lookup["symbol"],
name = tick_lookup["name"],
exchange = tick_lookup["exchange"],
exchangeToken = tick_lookup["exchangeToken"],
broker = "zerodhaKite",
playground/socketio/kafka_passthrough.py
+154
View File
@@ -0,0 +1,154 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Saturday, 28th Dec., 2024
OBJECTIVE:
To provide a quick way to test out passthrough messages over Kafka.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level:
import os
# Utils:
from utils_v2.string import json
from utils_v2.system import files
from utils_v2.date_time import date_time
from utils_v2.queue.async_kafka import ProducerKafka, ConsumerKafka, get_ssl_context
# For async activities:
import asyncio
# Common:
from shared import constants
# For random choices:
import random
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
# Define the test params:
TOPIC = "socket-io-bcast"
BOOTSTRAP_SERVERS = "del.ditscentre.in:9092"
cwd = files.get_cwd()
pdir = files.get_parent_directory(cwd, depth = 2)
SSL_CONTEXT = ssl_context = get_ssl_context(
ca_file = os.path.join(pdir, "creds", "kafka", "cert_authority.pem"),
cert_file = os.path.join(pdir, "creds", "kafka", "fullchain.pem"),
key_file = os.path.join(pdir, "creds", "kafka", "privkey.pem")
)
async def keep_producing():
# Create the producer:
my_producer = ProducerKafka(
topic = TOPIC,
bootstrap_servers = BOOTSTRAP_SERVERS,
security_protocol = "SSL",
ssl_context = SSL_CONTEXT
)
# Create a message for the producer to produce:
strategy_message = {
"to": "FC9O3N75Ax7B1v4NAAAD",
"event": "Strategy",
"namespace": "/finstitutions/trading",
"data": {
"message": "Your strategy ABC says buy XYZ.",
"playSound": True
}
}
corporate_action_message = {
"to": "FC9O3N75Ax7B1v4NAAAD",
"event": "Corporate Action",
"namespace": "/finstitutions/trading",
"data": {
"message": "Stock ABC has a corporate action tomorrow.",
"playSound": True,
"date": "29-Dec-2024",
"action": "Extraordinary Board Meeting"
}
}
# Keep sending the message in intervals:
while True:
producer_message = random.choice([
strategy_message,
corporate_action_message
])
success = await my_producer.produce(producer_message)
print("PRODUCED:", success)
await asyncio.sleep(1.0)
async def main():
await asyncio.gather(*[keep_producing()])
asyncio.run(main())
playground/socketio/to_kafka.py
+11 -2
View File
@@ -130,6 +130,8 @@ def to_kafka(tick: TradingTick) -> bool:
success = False
summary = tick.summary
print(json.to_string(summary))
print(json.to_string(tick.model_dump(), default=str))
summary["messageType"] = "ticks"
success = kafka_producer.produce(value = summary)
if not success:
@@ -203,8 +205,15 @@ def main():
# instruments += kite.instruments(exchange = "BCD")
# # Pick the instruments of interest:
instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments[:1000]]
# instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments if i["tradingsymbol"] in symbols_of_interest]
# instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments[:1000]]
# instruments = [
# TradingSymbol.from_zerodha_kite(i) for i in instruments
# if i["tradingsymbol"] in symbols_of_interest or i["name"] in symbols_of_interest
# ]
instruments = [
TradingSymbol.from_zerodha_kite(i) for i in instruments
if i["instrument_token"] in [109760007]
]
# Create the lookup:
for i in instruments:
wsio/finstitutions/trading/main.py
+120 -12
View File
@@ -106,8 +106,11 @@ EVENT_TICKS = "ticks"
# *****************************************************************************************************************
# For locking user-noting operations:
lock = asyncio.Semaphore(1)
# Session-awareness:
pass
connected_clients = {}
# Script-local:
flags = {
@@ -123,13 +126,20 @@ flags = {
@sio.on(event = EVENT_CONNECT, namespace = NAMESPACE_MODULE)
async def handle_connect(sid, environ):
async def handle_connect(sid, environ) -> bool:
# Start the common background processes:
if not flags.get("initDone"):
flags["initDone"] = True
asyncio.create_task(init())
# Note down user changes:
async with lock:
connected_clients[sid] = {
"user": None,
"rooms": []
}
# Allow/reject requests:
printer(sid)
print("ENVIRON:", json.to_string(environ, default = str))
@@ -140,7 +150,7 @@ async def handle_connect(sid, environ):
@sio.on(event = EVENT_DISCONNECT, namespace = NAMESPACE_MODULE)
async def handle_disconnect(sid, reason):
async def handle_disconnect(sid, reason) -> None:
printer(sid, reason)
@@ -148,7 +158,7 @@ async def handle_disconnect(sid, reason):
@sio.on(event = EVENT_ECHO, namespace = NAMESPACE_MODULE)
async def handle_echo(sid, data):
async def handle_echo(sid, data) -> None:
"""
For testing. This is a quick way to check if the module is up.
@@ -168,6 +178,88 @@ async def handle_echo(sid, data):
# ---------------------------------------------------------------------------------------------------------------------
async def send_passthrough(
to: str | List[str],
event: str,
namespace: str,
data: dict | list
) -> None:
"""
To send out the arbitrary passthrough message
:param to: The recipient of the message. This can be set to the 'sid' of a client to address only that client, or to
any custom room created by the application to address all the clients in that room, or to a list of custom
room names. If null, the event is broadcasted to all connected clients.
:param event: Any name for the event that the recipients are listening to. The strings 'connect', 'disconnect', and
'message' are reserved. Everything else is fair game.
:param namespace: The namespace (path) to send the data to.
:param data: The data to send to the target recipients.
:return: None.
"""
# Send out the event:
try: await sio.emit(
event = event,
data = data,
to = to,
namespace = namespace
)
except Exception as exception:
printer(exception)
# ---------------------------------------------------------------------------------------------------------------------
async def passthrough_from_kafka(
consumer: ConsumerKafka,
fetch_count: int = 100,
fetch_timeout: float = 1.0
) -> None:
"""
This function must run in the background forever and just keep listening for any passthrough messages from the
backend. The backend message must give the following kind of JSON:
{
"to": <sid>,
"event": <event-name>,
"namespace": <path>,
"data": <json-data>
}
:param consumer: The preconfigured Kafka consumer that can listen for ticks in asynchronous mode.
:param fetch_count: How many messages to consume in one go.
:param fetch_timeout: How long to wait (in seconds) while consuming messages from Kafka.
:return: None
"""
# Do the next part infinitely:
while True:
# Get messages form Kafka:
messages = await consumer.consume(
count = fetch_count,
timeout = fetch_timeout
)
# If there are no updates to give:
if not messages: continue
# Each message is a passthrough to be sent to the connected clients:
tasks = [
send_passthrough(
to = message["value"].get("to", None),
event = message["value"].get("event", None),
namespace = message["value"].get("namespace", "/"),
data = message["value"].get("data", {})
) for message in messages
]
results = await asyncio.gather(*tasks)
printer(len(messages))
# ---------------------------------------------------------------------------------------------------------------------
async def send_ticks(ticks: List[dict]):
"""
@@ -238,6 +330,14 @@ async def init():
# Start consuming ticks in the background:
cwd = files.get_cwd()
parent_dir = cwd
ssl_context = get_ssl_context(
ca_file = "/etc/ssl/dbu/ca.pem",
cert_file = "/etc/ssl/dbu/fullchain.pem",
key_file = "/etc/ssl/dbu/privkey.pem"
# ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"),
# cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"),
# key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem")
)
sio.start_background_task(
ticks_from_kafka,
consumer = ConsumerKafka(
@@ -245,19 +345,27 @@ async def init():
# group_id = f"{SERVER_HOSTNAME}_tickers",
bootstrap_servers = "del.ditscentre.in:9092",
security_protocol = "SSL",
ssl_context = get_ssl_context(
ca_file = "/etc/ssl/dbu/ca.pem",
cert_file = "/etc/ssl/dbu/fullchain.pem",
key_file = "/etc/ssl/dbu/privkey.pem"
# ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"),
# cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"),
# key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem")
),
ssl_context = ssl_context,
auto_offset_reset = "latest"
),
fetch_count = 1_250,
fetch_timeout = 1.0
)
sio.start_background_task(
passthrough_from_kafka,
consumer = ConsumerKafka(
topic = "socket-io-bcast",
# group_id = f"{SERVER_HOSTNAME}_tickers",
bootstrap_servers = "del.ditscentre.in:9092",
security_protocol = "SSL",
ssl_context = ssl_context,
auto_offset_reset = "latest"
),
fetch_count = 100,
fetch_timeout = 1.0
)
printer("Initialized.")
# *****************************************************************************************************************