🧹 chore: Updated files to comply with pylint requirements

Author: DevaOnBreaches

api/v1/alert.py

@@ -2,7 +2,6 @@
 
 # Standard library imports
 from datetime import datetime
-import sys
 import time
 
 # Third-party imports
@@ -14,7 +13,6 @@
 from user_agents import parse
 
 # Local imports
-from utils.custom_limiter import custom_rate_limiter
 from models.responses import (
     AlertResponse,
     UnsubscribeResponse,
@@ -23,6 +21,7 @@
 from services.analytics import get_breaches_analytics
 from services.breach import get_breaches, get_exposure, get_sensitive_exposure
 from services.send_email import send_alert_confirmation, send_unsub_email
+from utils.custom_limiter import custom_rate_limiter
 from utils.helpers import fetch_location_by_ip, get_preferred_ip_address
 from utils.token import confirm_token, generate_confirmation_token
 from utils.validation import validate_email_with_tld, validate_url, validate_variables

api/v1/analytics.py

@@ -5,44 +5,41 @@
 import html
 import logging
 from collections import defaultdict
-from typing import Optional, Union, Dict, Any, List
+from typing import Any, Dict, Optional, Union
 
 # Third-party imports
-from fastapi import APIRouter, HTTPException, Request, Query
+from fastapi import APIRouter, HTTPException, Query, Request
 from fastapi.responses import HTMLResponse, JSONResponse
 from fastapi.templating import Jinja2Templates
 from google.cloud import datastore
 from google.api_core import exceptions as google_exceptions
 
 # Local imports
-from utils.custom_limiter import custom_rate_limiter
 from models.responses import (
+    BreachDetails,
+    BreachHierarchyResponse,
+    DetailedBreachInfo,
     DetailedMetricsResponse,
-    PulseNewsResponse,
-    DomainAlertResponse,
     DomainAlertErrorResponse,
-    DomainBreachesResponse,
+    DomainAlertResponse,
     DomainBreachesErrorResponse,
-    BreachDetails,
-    DetailedBreachInfo,
-    ShieldActivationResponse,
+    DomainBreachesResponse,
+    PulseNewsResponse,
     ShieldActivationErrorResponse,
-    BreachHierarchyResponse,
-)
-from services.analytics import (
-    get_detailed_metrics,
-    get_pulse_news,
+    ShieldActivationResponse,
 )
+from services.analytics import get_detailed_metrics, get_pulse_news
 from services.send_email import send_dashboard_email_confirmation, send_shield_email
+from utils.custom_limiter import custom_rate_limiter
+from utils.helpers import fetch_location_by_ip, get_preferred_ip_address
+from utils.request import get_client_ip, get_user_agent_info
+from utils.token import confirm_token, generate_confirmation_token
 from utils.validation import (
     validate_email_with_tld,
+    validate_token,
     validate_url,
     validate_variables,
-    validate_token,
 )
-from utils.token import generate_confirmation_token, confirm_token
-from utils.helpers import get_preferred_ip_address, fetch_location_by_ip
-from utils.request import get_client_ip, get_user_agent_info
 
 router = APIRouter()
 templates = Jinja2Templates(directory="templates")

api/v1/api_keys.py

@@ -4,12 +4,13 @@
 import datetime
 import logging
 from typing import Optional
+
 from fastapi import APIRouter, Request
 from google.cloud import datastore
 
-from utils.custom_limiter import custom_rate_limiter
 from models.base import BaseResponse
-from utils.validation import validate_variables, validate_url
+from utils.custom_limiter import custom_rate_limiter
+from utils.validation import validate_url, validate_variables
 
 router = APIRouter()
 

api/v1/breaches.py

@@ -2,18 +2,16 @@
 
 # Standard library imports
 import json
-import sys
 from datetime import datetime
-from typing import Optional, Union, List, Dict
+from typing import Optional, Union
 
 # Third-party imports
-from fastapi import APIRouter, HTTPException, Request, Header, Query, Path
+from fastapi import APIRouter, Header, HTTPException, Path, Query, Request
 from fastapi.responses import JSONResponse, Response
 from google.cloud import datastore
 
 # Local imports
 from config.settings import MAX_EMAIL_LENGTH
-from utils.custom_limiter import custom_rate_limiter
 from models.responses import (
     BreachAnalyticsResponse,
     BreachAnalyticsV2Response,
@@ -28,13 +26,14 @@
     get_ai_summary,
     get_summary_and_metrics,
 )
-from services.breach import get_exposure, get_sensitive_exposure, get_breaches
+from services.breach import get_breaches, get_exposure, get_sensitive_exposure
+from utils.custom_limiter import custom_rate_limiter
 from utils.helpers import (
+    get_client_ip,
     string_to_boolean,
     validate_domain,
     validate_email_with_tld,
     validate_url,
-    get_client_ip,
 )
 from utils.validation import validate_variables
 

api/v1/domain_breaches.py

@@ -5,16 +5,14 @@
 from typing import Dict, List
 from operator import itemgetter
 
-from fastapi import APIRouter, Request, HTTPException, Header, Depends
+from fastapi import APIRouter, Depends, Header, HTTPException, Request
 from google.cloud import datastore
 from pydantic import BaseModel, Field
 
-from utils.custom_limiter import custom_rate_limiter
-from models.responses import DomainBreachesResponse, DomainBreachesErrorResponse
-
-# from services.analytics import get_domain_breach_analytics
-from utils.validation import validate_url  # , validate_api_key
 from models.base import BaseResponse
+from models.responses import DomainBreachesResponse
+from utils.custom_limiter import custom_rate_limiter
+from utils.validation import validate_url
 
 router = APIRouter()
 

api/v1/domain_phishing.py

@@ -1,23 +1,21 @@
 """Domain phishing check router module."""
 
-import socket
 import json
-import os
+import socket
 from datetime import datetime, timedelta
-from typing import Dict, List, Optional, Any, Union
 from pathlib import Path
+from typing import Any, Dict, List, Optional, Union
 
 import dnstwist
-from fastapi import APIRouter, Depends, HTTPException, Request, Query, Header
-from pydantic import BaseModel, Field, validator, EmailStr
+from fastapi import APIRouter, HTTPException, Query, Request
+from google.cloud import datastore
+from pydantic import BaseModel, EmailStr, Field, validator
 from redis import Redis
 
-from config.settings import REDIS_HOST, REDIS_PORT, REDIS_DB
+from config.settings import REDIS_DB, REDIS_HOST, REDIS_PORT
 from models.responses import BaseResponse
-from utils.validation import validate_email_with_tld, validate_variables
-from utils.token import confirm_token
-from google.cloud import datastore
 from utils.custom_limiter import custom_rate_limiter
+from utils.token import confirm_token
 
 router = APIRouter()
 
@@ -37,6 +35,7 @@
 
 
 def validate_file_content(file_path: Path, min_lines: int = 1) -> None:
+    """Validate the content of a file to ensure it has sufficient lines."""
     if not file_path.exists():
         raise FileNotFoundError(f"File not found: {file_path}")
     with open(file_path, "r") as f:
@@ -55,10 +54,13 @@ def validate_file_content(file_path: Path, min_lines: int = 1) -> None:
 
 
 class DomainPhishingRequest(BaseModel):
+    """Request model for domain phishing check."""
+
     domain: str = Field(..., description="Domain to check for phishing variants")
 
     @validator("domain")
     def validate_domain(cls, v: str) -> str:
+        """Validate the domain format and length."""
         if not v or not isinstance(v, str):
             raise ValueError("Domain must be a non-empty string")
         if len(v) > 255:
@@ -69,13 +71,17 @@ def validate_domain(cls, v: str) -> str:
 
 
 class DomainPhishingSummaryResponse(BaseResponse):
+    """Summary response model for domain phishing check."""
+
     total_scanned: int
     total_live: int = 0
     unique_fuzzers: int = 0
     last_checked: Optional[str] = None
 
 
 class DomainPhishingResponse(BaseResponse):
+    """Detailed response model for domain phishing check."""
+
     total_scanned: int
     total_live: int = 0
     unique_fuzzers: int = 0
@@ -85,6 +91,7 @@ class DomainPhishingResponse(BaseResponse):
 
 
 def is_domain_live(domain: str) -> bool:
+    """Check if a domain is live by resolving its IP address."""
     try:
         socket.gethostbyname(domain)
         return True
@@ -93,6 +100,7 @@ def is_domain_live(domain: str) -> bool:
 
 
 def get_cached_result(domain: str) -> Optional[Dict]:
+    """Retrieve cached phishing check results from Redis."""
     cache_key = f"phishing_check:{domain}"
     cached_data = redis_client.get(cache_key)
     if cached_data:
@@ -104,13 +112,15 @@ def get_cached_result(domain: str) -> Optional[Dict]:
 
 
 def cache_result(domain: str, result: Dict, expiry_hours: int = 24) -> None:
+    """Cache phishing check results in Redis."""
     cache_key = f"phishing_check:{domain}"
     if isinstance(result.get("last_checked"), datetime):
         result["last_checked"] = result["last_checked"].isoformat()
     redis_client.setex(cache_key, timedelta(hours=expiry_hours), json.dumps(result))
 
 
 async def verify_user_access(email: str, token: str) -> bool:
+    """Verify user access based on email and token."""
     if not email or not token:
         return False
     try:

api/v1/slack.py

@@ -2,15 +2,15 @@
 
 from fastapi import APIRouter, HTTPException, Request
 
-from utils.custom_limiter import custom_rate_limiter
 from models.requests import ChannelSetupRequest
 from models.responses import ChannelConfigResponse, ChannelSetupResponse
 from services.slack import (
-    setup_slack_channel,
-    verify_slack_channel,
     delete_slack_channel,
     get_slack_channel_config,
+    setup_slack_channel,
+    verify_slack_channel,
 )
+from utils.custom_limiter import custom_rate_limiter
 from utils.helpers import validate_url, validate_variables
 
 router = APIRouter()

config/middleware.py

@@ -6,7 +6,6 @@
 # Third-party imports
 from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
-from slowapi.middleware import SlowAPIMiddleware
 
 # Local imports
 from services.globe import process_request_for_globe

services/globe.py

@@ -5,12 +5,13 @@
 retrieves geolocation information, and publishes the data to a PubSub topic.
 """
 
+import hashlib
 import json
-import time
 import os
-import hashlib
+import time
+from typing import Any, Dict, Optional
+
 import httpx
-from typing import Dict, Any, Optional
 from google.cloud import pubsub_v1
 
 # Initialize PubSub constants

services/send_email.py

@@ -11,7 +11,6 @@
 
 # Third-party imports
 import httpx
-import requests
 from fastapi import HTTPException
 
 # Mailjet configuration