small bug fix

src/monitor_utils/dsql_shortlinks.py

@@ -0,0 +1,97 @@
+import asyncio
+from typing import Dict
+
+import httpx
+
+"""
+Minimal usage:
+
+    from dsql_shortlinks import generate_short_link_sync
+
+    result = generate_short_link_sync(
+        "https://example.com/product",
+        api_key="YOUR_DSQL_KEY",
+    )
+
+    print(result["short_url"])
+"""
+
+
+DEFAULT_BASE_URL = "https://api.profitlounge.us/link/affiliate"
+
+
+class DsqlShortlinks:
+    def __init__(
+        self,
+        api_key: str,
+        *,
+        base_url: str = DEFAULT_BASE_URL,
+        timeout: int = 60,
+    ) -> None:
+        self.api_key = api_key
+        self.base_url = base_url
+        self.timeout = timeout
+
+    async def create_short_link(self, url: str) -> Dict[str, str]:
+        headers = {"x-api-key": self.api_key}
+        payload = {"url": url}
+
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.post(
+                    self.base_url,
+                    headers=headers,
+                    json=payload,
+                    timeout=self.timeout,
+                )
+                response.raise_for_status()
+                data = response.json()
+                if isinstance(data, dict) and data.get("short_url"):
+                    return data
+        except Exception:
+            pass
+
+        return {"short_url": url, "long_url": url}
+
+    def create_short_link_sync(self, url: str) -> Dict[str, str]:
+        try:
+            running_loop = asyncio.get_running_loop()
+        except RuntimeError:
+            running_loop = None
+
+        if running_loop and running_loop.is_running():
+            raise RuntimeError(
+                "create_short_link_sync() cannot run inside an active event loop. "
+                "Use await create_short_link(...) instead."
+            )
+
+        return asyncio.run(self.create_short_link(url))
+
+
+async def generate_short_link(
+    url: str, api_key: str, timeout: int = 60
+) -> Dict[str, str]:
+    client = DsqlShortlinks(api_key, timeout=timeout)
+    return await client.create_short_link(url)
+
+
+def generate_short_link_sync(
+    url: str, api_key: str, timeout: int = 60
+) -> Dict[str, str]:
+    client = DsqlShortlinks(api_key, timeout=timeout)
+    return client.create_short_link_sync(url)
+
+
+if __name__ == "__main__":
+    EXAMPLE_URL = "https://woot.com/"
+    EXAMPLE_API_KEY = "YOUR_DSQL_KEY"
+
+    print("Example:")
+    print("from dsql_shortlinks import generate_short_link_sync")
+    print()
+    print("result = generate_short_link_sync(")
+    print(f'    "{EXAMPLE_URL}",')
+    print(f'    api_key="{EXAMPLE_API_KEY}",')
+    print(")")
+    print()
+    print('print(result["short_url"])')