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

(20241209) Standardizing the data models for the database. Started with mail authentication.

Browse Source
This commit is contained in:
khushal
2024-12-09 12:05:12 +05:30
parent 30088dcea4
commit e2c9200941
11 changed files with 262 additions and 244 deletions
Download Patch File Download Diff File Expand all files Collapse all files
models/data/database/__init__.py → models/data/core/__init__.py
View File
models/data/database/auth_token.py → models/data/core/auth_token.py
+31 -6
View File
@@ -43,6 +43,9 @@ from typing import Optional, Literal, Union
from utils_v2.string import regex
from utils_v2.date_time import date_time
# Other core models:
from models.data.core.user_info import CoreUserInfoModel
# To work with MongoDB:
from bson.objectid import ObjectId
@@ -108,31 +111,41 @@ class CoreAuthTokenModel(BaseModel):
firstRequestTs: AwareDatetime = Field(
description = "the time (utc) at which authorization was first requested",
frozen = True
frozen = True,
default_factory = lambda: date_time.get_current_utc_date_time(as_string = False)
)
lastRequestTs: AwareDatetime = Field(
description = "the time (utc) at which authorization was last requested",
frozen = False
frozen = False,
default = None
)
firstRefreshTs: AwareDatetime = Field(
description = "the time (utc) at which the tokens were first refreshed",
frozen = False
frozen = False,
default = None
)
lastRefreshTs: AwareDatetime = Field(
description = "the time (utc) at which the tokens were last refreshed",
frozen = False
frozen = False,
default = None
)
auth: dict | None = Field(
description = "any direct auth details like api keys or passwords; will differ for each client",
frozen = True,
default = None
)
token: dict | None = Field(
description = "the actual auth tokens of that client; will differ for each client",
frozen = True,
default_factory = lambda: date_time.get_current_utc_date_time(as_string = False)
default = None
)
user: dict = Field(
user: CoreUserInfoModel = Field(
description = "how you identify your user",
frozen = True
)
@@ -142,6 +155,18 @@ class CoreAuthTokenModel(BaseModel):
frozen = True
)
status: Literal["pending", "active", "disabled"] = Field(
description = "to indicate the status of this account",
frozen = False,
default = "pending"
)
syncFreq: Literal[60, 300, 1500] = Field(
description = "the no. of seconds after which to poll for updates from the client (if applicable)",
frozen = False,
default = 300
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
models/data/database/message.py → models/data/core/message.py
View File
models/data/database/payment.py → models/data/core/payment.py
+92 -41
View File
@@ -37,7 +37,7 @@ sys.path.append("..")
# For making data behaviour_models:
from pydantic import BaseModel, Field, field_validator, PastDatetime, AwareDatetime
from typing import Optional, Literal, Union
from typing import Optional, Literal, Union, List
# My utils:
from utils_v2.string import regex
@@ -80,6 +80,50 @@ import pycountry
# *****************************************************************************************************************
class PaymentEvent(BaseModel):
eventTs: AwareDatetime = Field(
description = "to know the date and time (utc) of this update",
frozen = True
)
initByPG: bool = Field(
description = "to figure out whether the payment gateway initiated this event or we did",
frozen = True
)
httpCode: int | None = Field(
description = "the http code generated by the event",
frozen = True,
examples = [200, 400, 401]
)
payload: dict = Field(
description = "the json payload or set of query params received from an event from the payment gateway",
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "allow"
arbitrary_types_allowed = True
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator("eventTs", mode = "before")
def parse_date_time(cls, value):
return date_time.parse_date_time(input_value = value, timezone = date_time.TIMEZONE_UTC)
# ---------------------------------------------------------------------------------------------------------------------
class CorePaymentModel(BaseModel):
version: str = Field(
@@ -89,22 +133,22 @@ class CorePaymentModel(BaseModel):
default = "1.0.0"
)
paymentStatus: Literal["requested", "paid", "rejected"] = Field(
paymentStatus: Literal[
"initFailed", # ... When we tried to initiate the request, but the payment gateway (PG) rejected it.
"initiated", # .... When we made a successful payment request, or the customer initiated one from the PG.
"failed", # ....... When the customer tried paying, but it failed (e.g.: because of an incorrect pin).
"rejected", # ..... When the customer explicitly rejected the payment.
"authorized", # ... When the customer made the payment (but it hasn't been settled in your account yet).
"settled", # ...... When the PG sends the money to your account.
"refunded", # ..... When the money was refunded to the client.
] = Field(
description = "the status of the payment request to see what stage of the process we are in",
frozen = False,
default = "requested"
frozen = False
)
requestTs: AwareDatetime = Field(
description = "the time (utc) at which the payment request was initiated",
frozen = True,
default_factory = lambda: date_time.get_current_utc_date_time(as_string = False)
)
responseTs: AwareDatetime | None = Field(
description = "the time (utc) at which the payer responded to the payment request",
frozen = False,
default = None
lastEventTs: AwareDatetime = Field(
description = "the time (utc) at which the latest payment event occurred",
frozen = True
)
tokenId: ObjectId = Field(
@@ -122,7 +166,7 @@ class CorePaymentModel(BaseModel):
examples = ["INR", "USD", "KES"]
)
metadata: dict = Field(
metadata: dict | None = Field(
description = "any arbitrary amount of data to identify the user and payment details",
frozen = True
)
@@ -138,28 +182,9 @@ class CorePaymentModel(BaseModel):
default = None
)
clientPaymentRequestHttpCode: int | str = Field(
description = "the http code the third-party client returned when you requested the payment",
frozen = False,
default = None
)
clientPaymentRequestJSON: str | int = Field(
description = "how the third-party client responded when you requested the payment",
frozen = False,
default = None
)
clientPaymentResponseHttpCode: int | str = Field(
description = "the http code the third-party client returned when your user responded to the payment request",
frozen = False,
default = None
)
clientPaymentResponseJSON: str | int | None = Field(
description = "how the third-party client responded when your user responded to the payment request",
frozen = False,
default = None
events: List[PaymentEvent] = Field(
description = "an array of all the events that happened in the process of this payment",
frozen = False
)
# ┏┓ ┏•
@@ -175,7 +200,7 @@ class CorePaymentModel(BaseModel):
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator("requestTs", "responseTs", mode = "before")
@field_validator("lastEventTs", mode = "before")
def parse_date_time(cls, value):
return date_time.parse_date_time(input_value = value, timezone = date_time.TIMEZONE_UTC)
@@ -203,16 +228,42 @@ if __name__ == "__main__":
from utils_v2.string import json
now = date_time.get_current_utc_date_time(as_string = False)
payment = CorePaymentModel(
paymentStatus = "requested",
paymentStatus = "authorized",
tokenId = "67519cf3a7804fcbc6f12452",
amount = 1.00,
currencyCode = "INR",
metadata = {
"userId": 1,
"name": "My Test"
"name": "Bhopli"
},
client = "safaricomMPesaExpress"
client = "razorpay",
clientPaymentReferenceId = "txn_123_abc",
lastEventTs = now,
events = [
PaymentEvent(
eventTs = now - datetime.timedelta(minutes = 1, seconds = 12),
initByPG = True,
httpCode = None,
payload = {
"status": "captured",
"from": "Barfi",
}
),
PaymentEvent(
eventTs = now,
initByPG = True,
httpCode = None,
payload = {
"status": "authorized",
"from": "Barfi",
"amount": -100.00,
"description": "meow"
}
)
]
)
print("PAYMENT TXN. MODEL:", json.to_string(payment.model_dump(), default = str))
models/data/core/user_info.py
+28 -97
View File
@@ -6,11 +6,11 @@
DATE:
Saturday, 7th Dec., 2024.
Monday, 9th Dec., 2024.
OBJECTIVE:
To define how auth tokens will be stored in the database.
To define how user info will be stored in the database.
REFERENCES:
@@ -77,7 +77,7 @@ import datetime
# *****************************************************************************************************************
class CoreAuthTokenModel(BaseModel):
class CoreUserInfoModel(BaseModel):
version: str = Field(
description = "a hint about the version no. of this message",
@@ -86,72 +86,40 @@ class CoreAuthTokenModel(BaseModel):
default = "1.0.0"
)
serviceType: Literal["email", "sms", "chat"] = Field(
description = "the kind of service this message was sent/received from",
frozen = True
)
client: Literal[
"gmail", "outlook", # ...................... Mail Clients
"telegram", "whatsapp", # .................. Chat Clients
"nimbusSmsIndia", "savvyBulkSmsKenya", # ... SMS Clients
"razorpay", "safaricomMPesaExpress" # ...... Payment Gateways
] = Field(
description = "the third-part client that was used",
frozen = True
)
authType: Literal["oauth", "auth"] = Field(
description = "the type of authentication procedure used",
frozen = True
)
firstRequestTs: AwareDatetime = Field(
description = "the time (utc) at which authorization was first requested",
frozen = True
)
lastRequestTs: AwareDatetime = Field(
description = "the time (utc) at which authorization was last requested",
frozen = False
)
firstRefreshTs: AwareDatetime = Field(
description = "the time (utc) at which the tokens were first refreshed",
frozen = False
)
lastRefreshTs: AwareDatetime = Field(
description = "the time (utc) at which the tokens were last refreshed",
frozen = False
)
token: dict | None = Field(
description = "the actual auth tokens of that client; will differ for each client",
fullName: str | None = Field(
description = "the full name of the user as found in the database",
frozen = True,
default_factory = lambda: date_time.get_current_utc_date_time(as_string = False)
examples = ["Bhopli Narangi"]
)
user: dict = Field(
description = "how you identify your user",
userId: int | str | None = Field(
description = "the id of the user as found in the database",
frozen = True
)
clientUserId: dict = Field(
description = "how third-party client identifies the same user",
entityId: int | str | None = Field(
description = "the id of the entity with which this user is associated",
frozen = True
)
status: Literal["active", "disabled"] = Field(
description = "to indicate the status of this account",
frozen = False,
default = "active"
billingAccountId: int | str | None = Field(
description = "the id of the billing account with which this user is associated",
frozen = True
)
syncFreq: Literal[60, 300, 1500] = Field(
description = "the no. of seconds after which to poll for updates from the client (if applicable)",
frozen = False,
default = 300
departmentId: int | str | None = Field(
description = "the id of the dept. in which this user is working",
frozen = True
)
branchId: int | str | None = Field(
description = "the id of the branch in which this user is working",
frozen = True
)
industry: str | None = Field(
description = "the name of the industry this user is working in",
frozen = True
)
# ┏┓ ┏•
@@ -160,21 +128,9 @@ class CoreAuthTokenModel(BaseModel):
# ┛
class Config:
extra = "allow"
extra = "ignore"
arbitrary_types_allowed = True
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator(
"firstRequestTs",
"lastRequestTs", "firstRefreshTs", "lastRefreshTs",
mode = "before"
)
def parse_date_time(cls, value):
return date_time.parse_date_time(input_value = value, timezone = date_time.TIMEZONE_UTC)
# *****************************************************************************************************************
# ***** ****
@@ -184,29 +140,4 @@ class CoreAuthTokenModel(BaseModel):
if __name__ == "__main__":
from utils_v2.string import json
auth_token = CoreAuthTokenModel(
serviceType = "email",
client = "gmail",
authType = "oauth",
firstRequestTs = date_time.get_current_utc_date_time(as_string = False),
lastRequestTs = date_time.get_current_utc_date_time(as_string = False),
firstRefreshTs = date_time.get_current_utc_date_time(as_string = False),
lastRefreshTs = date_time.get_current_utc_date_time(as_string = False),
token = {
"username": "testing123",
"password": "abcdefgh"
},
user = {
"userId": 0,
"entityId": 1,
"billingAccountId": 2,
"fullName": "Bhopli"
},
clientUserId = {
"email": "bhopli@gmail.com"
}
)
print("AUTH-TOKEN MODEL:", json.to_string(auth_token.model_dump(), default = str))
pass