yatmesh
65 lines
2.2 KiB
Python
65 lines
2.2 KiB
Python
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()) |