Merge 907ead0 into 6b46e9c

abuse_whois/__init__.py

@@ -1,18 +1,4 @@
-from typing import Optional
-
-from abuse_whois.matchers.shared_hosting import get_shared_hosting_provider
-from abuse_whois.matchers.whois import get_contact_from_whois
-
-from .errors import InvalidAddressError
-from .ip import resolve_ip_address
-from .schemas import Contact, Contacts
-from .utils import (
-    get_hostname,
-    get_registered_domain,
-    is_domain,
-    is_ip_address,
-    is_supported_address,
-)
+from .main import get_abuse_contacts  # noqa: F401
 
 try:
     import importlib.metadata as importlib_metadata
@@ -21,44 +7,3 @@
 
 
 __version__ = importlib_metadata.version(__name__)
-
-
-async def get_abuse_contacts(address: str) -> Contacts:
-    if not is_supported_address(address):
-        raise InvalidAddressError(f"{address} is not supported type address")
-
-    shared_hosting_provider: Optional[Contact] = None
-    registrar: Optional[Contact] = None
-    hosting_provider: Optional[Contact] = None
-
-    hostname = get_hostname(address)
-    ip_address: Optional[str] = None
-    registered_domain: Optional[str] = None
-
-    shared_hosting_provider = get_shared_hosting_provider(hostname)
-
-    if is_domain(hostname):
-        registered_domain = get_registered_domain(hostname)
-        registrar = await get_contact_from_whois(hostname)
-
-        # get IP address by domain
-        try:
-            ip_address = await resolve_ip_address(hostname)
-        except OSError:
-            pass
-
-    if is_ip_address(hostname):
-        ip_address = hostname
-
-    if ip_address is not None:
-        hosting_provider = await get_contact_from_whois(ip_address)
-
-    return Contacts(
-        address=address,
-        hostname=hostname,
-        ip_address=ip_address,
-        registered_domain=registered_domain,
-        shared_hosting_provider=shared_hosting_provider,
-        registrar=registrar,
-        hosting_provider=hosting_provider,
-    )

abuse_whois/main.py

@@ -0,0 +1,76 @@
+from functools import partial
+from typing import Optional
+
+import aiometer
+
+from abuse_whois.matchers.shared_hosting import get_shared_hosting_provider
+from abuse_whois.matchers.whois import get_contact_from_whois
+
+from .errors import InvalidAddressError
+from .ip import resolve_ip_address
+from .schemas import Contact, Contacts
+from .utils import (
+    get_hostname,
+    get_registered_domain,
+    is_domain,
+    is_ip_address,
+    is_supported_address,
+)
+
+
+async def get_contact(domain_or_ip_address: Optional[str] = None) -> Optional[Contact]:
+    if domain_or_ip_address is None:
+        return None
+
+    return await get_contact_from_whois(domain_or_ip_address)
+
+
+async def get_registrar_and_hosting_provider_contacts(
+    *, domain: Optional[str] = None, ip_address: Optional[str] = None
+):
+    values = [domain, ip_address]
+    return await aiometer.run_all([partial(get_contact, value) for value in values])
+
+
+async def get_abuse_contacts(address: str) -> Contacts:
+    if not is_supported_address(address):
+        raise InvalidAddressError(f"{address} is not supported type address")
+
+    shared_hosting_provider: Optional[Contact] = None
+    registrar: Optional[Contact] = None
+    hosting_provider: Optional[Contact] = None
+
+    hostname = get_hostname(address)
+
+    domain: Optional[str] = None  # FQDN
+    ip_address: Optional[str] = None
+    registered_domain: Optional[str] = None
+
+    shared_hosting_provider = get_shared_hosting_provider(hostname)
+
+    if is_domain(hostname):
+        domain = hostname
+        registered_domain = get_registered_domain(hostname)
+
+        # get IP address by domain
+        try:
+            ip_address = await resolve_ip_address(hostname)
+        except OSError:
+            pass
+
+    if is_ip_address(hostname):
+        ip_address = hostname
+
+    registrar, hosting_provider = await get_registrar_and_hosting_provider_contacts(
+        domain=domain, ip_address=ip_address
+    )
+
+    return Contacts(
+        address=address,
+        hostname=hostname,
+        ip_address=ip_address,
+        registered_domain=registered_domain,
+        shared_hosting_provider=shared_hosting_provider,
+        registrar=registrar,
+        hosting_provider=hosting_provider,
+    )

pyproject.toml

@@ -10,21 +10,22 @@ readme = "README.md"
 
 [tool.poetry.dependencies]
 python = "^3.7"
+aiometer = "^0.3.0"
 asyncer = "^0.0.1"
 cachetools = "^5.2.0"
 email-validator = "^1.2.1"
 fastapi = "^0.78.0"
 gunicorn = { version = "^20.1.0", optional = true }
 importlib-metadata = {version = "^4.11.4", python = "<3.8"}
-typing-extensions = {version = "^4.2.0", python = "<3.8"}
 loguru = "^0.6.0"
-pydantic = "^1.9.0"
-pyhumps = "^3.7.1"
+pydantic = "^1.9.1"
+pyhumps = "^3.7.2"
 PyYAML = "^6.0"
 sh = "^1.14.2"
 tldextract = "^3.3.0"
 typer = "^0.4.1"
-uvicorn = { extras = ["standard"], version = "^0.17.6", optional = true }
+typing-extensions = {version = "^3.10", python = "<3.8"}
+uvicorn = { extras = ["standard"], version = "^0.18.1", optional = true }
 whois-parser = "^0.1.4"
 
 [tool.poetry.dev-dependencies]

tests/terst_version.py

@@ -0,0 +1,5 @@
+from abuse_whois import __version__
+
+
+def test_version():
+    assert isinstance(__version__, str)