(20250811) - Added RazorPay Integration class with auth api and requires validations and many more.

2025-08-11 14:19:24 +05:30
parent 1732e12adf
commit ddc29a7fe1
9 changed files with 723 additions and 7 deletions
@@ -0,0 +1,65 @@
+from utils_v2.goog.controllers.base import AsyncGoogleBase
+import httpx
+import asyncio
+from utils_v2.string import json
+
+
+# Create an HTTP client:
+test_client = httpx.AsyncClient(
+    limits = httpx.Limits(
+        max_connections = 100,  # ............ Maximum number of connections allowed in the pool.
+        max_keepalive_connections = 50,  # ... Maximum number of connections that can be kept alive.
+    ),
+    timeout = httpx.Timeout(
+        connect = 2.5,  # ... Shorter connection timeout.
+        read = 2.5,  # ...... Like what EasyEcom gives.
+        write = 10.0,  # .... Time to wait for sending data.
+        pool = 120.0  # ..... Time to wait for a free connection from the pool.
+    )
+)
+
+# Read the secrets that give you access to the app:
+secrets_file = r"/home/python-dev-debug/Downloads/client_secret_349360248417-uuba8eudk75jg1jag212g5obhc1uostk.apps.googleusercontent.com.json"
+secrets_dict = json.from_file(secrets_file)
+
+PLACES_SCOPES = ["https://www.googleapis.com/auth/cloud-platform", "https://www.googleapis.com/auth/userinfo.profile"]
+
+async def main():
+
+    # Create an instance of the client:
+    my_places = AsyncGoogleBase(
+        service_name = "places",
+        oauth_json = secrets_dict,
+        http_client = test_client,
+        # redirect_url = r"https://api.thecaoffice.com/converse/software/callback/places",
+        redirect_url = r"https://wtt.ditscentre.in/shopify/test/1",
+        debug = True,
+        debug_prefix = "places (M) | ",
+        debug_only_errors = False
+    )
+
+    # Request Auth:
+    print("AUTH URL:", await my_places.get_authorization_url(
+        scopes = PLACES_SCOPES,
+        state = "Sundar",
+        approval_prompt = "force"
+    ))
+
+    # Get tokens from callback:
+    test_tokens = await my_places.get_authorization_tokens(
+        scopes = PLACES_SCOPES,
+        redirect_url = input("Paste the redirect URL here: ")
+    )
+    print("TOKENS:", test_tokens)
+
+    # Test some feature:
+    # response = await my_gmail.get_user_profile(tokens = test_tokens)
+    # print("SUCCESS:", response.success)
+    # print("SUMMARY:", response.to_markdown())
+    # print("\n\n---\n\n")
+    # print("DATA:", json.to_string(response.data, default = str))
+    # if not response.success:
+    #     print("\n\n---\n\n")
+
+
+asyncio.run(main())