Merge pull request #49 from IKermani/async-support

Async check.

Makefile

@@ -11,7 +11,7 @@ lock: ## Update poetry.lock
 	poetry lock
 
 run: ## Run project
-	poetry run katran-master
+	poetry run check-filter
 
 build: clean ## Build package
 	poetry build

check_filter/__init__.py

@@ -5,3 +5,5 @@
 __version__ = "2.0.3"
 __author__ = "Arash Hatami <info@arash-hatami.ir>"
 __epilog__ = "Made with :heart:  in [green]Iran[/green]"
+
+from .check import DomainChecker

check_filter/check.py

@@ -1,28 +1,39 @@
 """It's the checker module for filtering status"""
 import os
-import dns.resolver
+import dns.asyncresolver
 
-blocked_ips = {'10.10.34.34', '10.10.34.35', '10.10.34.36'}
-resolver = dns.resolver.Resolver()
-resolver.nameservers = [
-    '178.22.122.100' if 'CI' in os.environ else '8.8.8.8'
-]
-headers = ['Address', 'Status']
 
+class DomainChecker:
+    """
+    Checks if the domain is blocked or not
+    """
+    headers = ['Address', 'Status']
 
-def check(domain: str) -> bool:
-    """Check if domain is blocked
+    def __init__(self, blocked_ips=None):
+        self.blocked_ips = blocked_ips or {
+            '10.10.34.34',
+            '10.10.34.35',
+            '10.10.34.36',
+        }
 
-    Args:
-        domain (str): The domain's name
+        self.resolver = dns.asyncresolver.Resolver(configure=False)
+        self.resolver.nameservers = [
+            '178.22.122.100' if 'CI' in os.environ else '8.8.8.8'
+        ]
 
-    Returns:
-        bool: Blocking status
-    """
-    try:
-        answer = resolver.resolve(domain)
-        ip_list = [data.address for data in answer]
-    except dns.resolver.NXDOMAIN:
-        ip_list = []
+    async def acheck(self, domain: str) -> tuple[str, bool]:
+        """Check if domain is blocked
+
+        Args:
+            domain (str): The domain's name
+
+        Returns:
+            tuple: (Domain's name, Blocking status)
+        """
+        try:
+            answer = await self.resolver.resolve(domain)
+            ip_list = [data.address for data in answer]
+        except dns.resolver.NXDOMAIN:
+            ip_list = []
 
-    return not any(ip in blocked_ips for ip in ip_list)
+        return domain, not any(ip in self.blocked_ips for ip in ip_list)

check_filter/cli.py

@@ -1,4 +1,5 @@
 """This module provides the CheckFilter CLI."""
+import asyncio
 
 from typing import Optional
 import typer
@@ -9,7 +10,6 @@
     __version__,
     __description__,
     __epilog__,
-    check,
     utils
 )
 
@@ -34,13 +34,12 @@ def domain(domain: str) -> None:
     Args:
         domain (str): Domain's name
     """
-    if utils.validate_domain(domain):
-        p(f"[yellow]Checking [italic]{domain}[/italic] ...[/yellow]")
-        result = check.check(domain=domain)
-        p(utils.print_result(domain=domain, result=result))
-        return
+    if not utils.validate_domain(domain):
+        p(f"[red]The `{domain}` is not a valid domain name![/red]")
+        raise typer.Exit()
 
-    p(f"[red]The `{domain}` is not a valid domain name![/red]")
+    p(f"[yellow]Checking [italic]{domain}[/italic] ...[/yellow]")
+    asyncio.run(utils.print_result([domain]))
 
 
 @app.command(epilog=__epilog__)
@@ -51,17 +50,15 @@ def domains(domains: str) -> None:
     Args:
         domains (str): A comma separated list of domains
     """
-    result = []
     p("[yellow]Checking domains ...[/yellow]")
+    domains = domains.split(",")
+    domain_validity_checks = [
+        utils.validate_domain(domain) for domain in domains
+    ]
+    if not all(domain_validity_checks):
+        raise typer.Exit()
 
-    for domain in domains.split(","):
-        if utils.validate_domain(domain):
-            status = check.check(domain=domain)
-            result.append([domain, status])
-        else:
-            p(f"[red]The `{domain}` is not a valid domain name![/red]")
-
-    utils.print_table(result)
+    asyncio.run(utils.print_result(domains))
 
 
 @app.command(epilog=__epilog__)
@@ -74,28 +71,31 @@ def file(path: str):
     """
     p("[yellow]Checking domains ...[/yellow]")
     with open(file=path, encoding="utf-8", mode='r') as file:
-        sites = [site.strip() for site in file]
-        result = []
+        domains = [domain.strip() for domain in file]
+
+    domain_validity_checks = [
+        utils.validate_domain(domain) for domain in domains
+    ]
+    if not all(domain_validity_checks):
+        raise typer.Exit()
 
-        for domain in sites:
-            if utils.validate_domain(domain):
-                status = check.check(domain=domain)
-                result.append([domain, status])
-            else:
-                p(f"[red]The `{domain}` is not a valid domain name![/red]")
-        utils.print_table(result)
+    asyncio.run(utils.print_result(domains))
 
 
 @app.callback()
 def main(
-    version: Optional[bool] = typer.Option(  # pylint: disable=unused-argument
-        None,
-        "--version",
-        "-v",
-        help="Show the application's version.",
-        callback=_version_callback,
-        is_eager=True,
-    )
+        version: Optional[bool] = typer.Option(  # pylint: disable=unused-argument
+            None,
+            "--version",
+            "-v",
+            help="Show the application's version.",
+            callback=_version_callback,
+            is_eager=True,
+        )
 ) -> None:
     """It's the version printer for CLI"""
     return
+
+
+if __name__ == "__main__":
+    app()

check_filter/utils.py

@@ -1,10 +1,13 @@
 """Utils module"""
+import asyncio
 
 import validators
+from rich import print as p
 from rich.console import Console
+from rich.live import Live
 from rich.table import Table
 
-console = Console()
+from check_filter import DomainChecker
 
 
 def validate_domain(domain: str) -> bool:
@@ -13,30 +16,17 @@ def validate_domain(domain: str) -> bool:
     Args:
         domain (str): Domain's name
 
-    Returns:
-        bool: Result of validation
+    Returns: is domain valid
     """
-    return validators.domain(domain)
+    is_valid = validators.domain(domain)
+    if not is_valid:
+        p(f"[red]The `{domain}` is not a valid domain name![/red]")
 
+    return is_valid
 
-def print_result(domain: str, result: bool) -> str:
-    """Print a pretty result for CLI
-
-    Args:
-        domain (str): Domain's name
-        result (bool): Blocking result
-
-    Returns:
-        str: Printable result
-    """
-    if result:
-        return f"\nThe `[italic]{domain}[/italic]` is [green]free[/green] in Iran :smiley:"
 
-    return f"\nThe `[italic]{domain}[/italic]` is [red]blocked[/red] in Iran :x:"
-
-
-def print_table(domains: list) -> None:
-    """Print a pretty table for CLI
+async def print_result(domains: list) -> None:
+    """Print a pretty result for CLI
 
     Args:
         domains (list): List of [domain, status] of blocking result
@@ -45,9 +35,20 @@ def print_table(domains: list) -> None:
     table.add_column("Domain", justify="left", no_wrap=True)
     table.add_column("Status", justify="left", no_wrap=True)
 
-    for domain in domains:
-        table.add_row(
-            f"[red]{domain[0]}[/red]" if not domain[1] else domain[0],
-            "[green]Free[/green]" if domain[1] else "[red]Blocked[/red] :x:")
+    domain_checker = DomainChecker()
+
+    tasks = set()
+
+    with Live(table, auto_refresh=False) as live_table:
+        for d in domains:
+            tasks.add(asyncio.create_task(
+                domain_checker.acheck(d),
+                name=f"domain-check-{d}",
+            ))
 
-    console.print("\n", table)
+        for future in asyncio.as_completed(tasks):
+            domain, status = await future
+            table.add_row(
+                f"[red]{domain}[/red]" if not status else domain,
+                "[green]Free[/green]" if status else "[red]Blocked[/red] :x:")
+            live_table.refresh()