From 0844e4935f8d49be28530787040d99dc9a1c7262 Mon Sep 17 00:00:00 2001
From: Dmitry <b4tm4n@mail.ru>
Date: Wed, 21 Feb 2024 09:42:38 +0300
Subject: [PATCH] add generic http source providers
---
pddnsc/sources/http.py | 75 ++++++++++++++++++++++++++++++++++++++++++
1 file changed, 75 insertions(+)
create mode 100644 pddnsc/sources/http.py
diff --git a/pddnsc/sources/http.py b/pddnsc/sources/http.py
new file mode 100644
index 0000000..ec65934
--- /dev/null
+++ b/pddnsc/sources/http.py
@@ -0,0 +1,75 @@
+import httpx
+
+from pddnsc.base import BaseSourceProvider, filter_ipv4, filter_ipv6
+
+
+class GenericHttpSource(BaseSourceProvider):
+ def post_init(self):
+ self.url_v4 = self.config.get("url_v4")
+ self.url_v6 = self.config.get("url_v6")
+ self.headers = self.config.get("headers", {})
+
+ async def fetch_v4(self) -> str:
+ if not self.url_v4:
+ return None
+
+ result = ""
+ async with httpx.AsyncClient(
+ transport=self.ipv4t, headers=self.headers
+ ) as client:
+ response = await client.get(self.url_v4)
+ if response.is_success:
+ result = response.text.strip()
+ return filter_ipv4(result)
+
+ async def fetch_v6(self) -> str:
+ if not self.url_v6:
+ return None
+
+ result = ""
+ async with httpx.AsyncClient(
+ transport=self.ipv6t, headers=self.headers
+ ) as client:
+ response = await client.get(self.url_v6)
+ if response.is_success:
+ result = response.text.strip()
+ return filter_ipv6(result)
+
+
+class GenericHttpJsonSource(BaseSourceProvider):
+ def post_init(self):
+ self.url_v4 = self.config.get("url_v4")
+ self.url_v6 = self.config.get("url_v6")
+ self.key_v4 = self.config.get("key_v4")
+ self.key_v6 = self.config.get("key_v6")
+ if self.config.get("use_accept", True):
+ headers = {"Accept": self.config.get("acept_type", "application/json")}
+ else:
+ headers = {}
+ self.headers = headers.update(self.config.get("headers", {}))
+
+ async def fetch_v4(self) -> str:
+ if not self.url_v4 or self.key_v4 is None:
+ return None
+
+ result = ""
+ async with httpx.AsyncClient(
+ transport=self.ipv4t, headers=self.headers
+ ) as client:
+ response = await client.get(self.url_v4)
+ if response.is_success:
+ result = response.json()[self.key_v4].strip()
+ return filter_ipv4(result)
+
+ async def fetch_v6(self) -> str:
+ if not self.url_v6 or self.key_v6 is None:
+ return None
+
+ result = ""
+ async with httpx.AsyncClient(
+ transport=self.ipv6t, headers=self.headers
+ ) as client:
+ response = await client.get(self.url_v6)
+ if response.is_success:
+ result = response.json()[self.key_v6].strip()
+ return filter_ipv6(result)