# Import necessary libraries import os import sqlite3 import json import requests from fastapi import FastAPI, HTTPException, Path, status from pydantic import BaseModel, Field # --- Define the structure of our API's successful response --- class NumberInfoResponse(BaseModel): phone_number: str = Field(..., example="+14155552671", description="The phone number in international format.") is_valid: bool = Field(..., example=True, description="Whether the phone number is valid.") country_code: str | None = Field(None, example="US", description="The two-letter country code.") location: str | None = Field(None, example="Novato", description="The general location of the number.") carrier: str | None = Field(None, example="AT&T Mobility LLC", description="The mobile carrier for the number.") line_type: str | None = Field(None, example="mobile", description="The type of line (e.g., mobile, landline).") # --- Application Configuration --- NUMVERIFY_API_KEY = os.getenv("NUMVERIFY_API_KEY") DATABASE_PATH = "/data/phone_data_cache.db" # Create the main FastAPI application app = FastAPI( title="Phone Number Intelligence API", description="A complete API to get information about a phone number, with a smart caching system.", version="1.1.0", # Version up! ) # --- Database Functions (No changes here) --- def init_database(): os.makedirs(os.path.dirname(DATABASE_PATH), exist_ok=True) with sqlite3.connect(DATABASE_PATH) as conn: cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS number_cache ( phone_number TEXT PRIMARY KEY, is_valid BOOLEAN, country_code TEXT, location TEXT, carrier TEXT, line_type TEXT, raw_response TEXT ) ''') print("SUCCESS: Database is ready at:", DATABASE_PATH) def get_number_from_cache(phone_number: str) -> dict | None: if not os.path.exists(DATABASE_PATH): return None with sqlite3.connect(DATABASE_PATH) as conn: conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute("SELECT * FROM number_cache WHERE phone_number = ?", (phone_number,)) data = cursor.fetchone() return dict(data) if data else None def save_number_to_cache(phone_number: str, data: dict): with sqlite3.connect(DATABASE_PATH) as conn: cursor = conn.cursor() cursor.execute(''' INSERT OR REPLACE INTO number_cache (phone_number, is_valid, country_code, location, carrier, line_type, raw_response) VALUES (?, ?, ?, ?, ?, ?, ?) ''', ( phone_number, data.get('valid'), data.get('country_code'), data.get('location'), data.get('carrier'), data.get('line_type'), json.dumps(data) )) print(f"SUCCESS: Saved {phone_number} to the cache.") # --- External API Call Function (IMPROVED!) --- def fetch_from_numverify(phone_number: str) -> dict: """ Fetches new data from the external API Layer service. This version includes much better error handling. """ if not NUMVERIFY_API_KEY: print("CRITICAL ERROR: 'NUMVERIFY_API_KEY' is not set.") # Raise an exception that FastAPI will turn into a proper JSON error response raise HTTPException(status_code=500, detail="API server is not configured correctly.") url = f"https://api.apilayer.com/number_verification/validate?number={phone_number}" headers = {"apikey": NUMVERIFY_API_KEY} try: response = requests.get(url, headers=headers, timeout=10) response.raise_for_status() # Check for HTTP errors like 500, 403, etc. data = response.json() # <<< CRITICAL IMPROVEMENT #1 >>> # API Layer can return a 200 OK status but with an error inside the JSON. # This happens when you run out of API credits. if data.get("success") is False: # Extract their error message to send to our user error_info = data.get("error", {}).get("info", "Unknown error from provider.") raise HTTPException(status_code=429, detail=f"API limit likely reached. Provider message: {error_info}") return data except requests.RequestException as e: print(f"Error calling external API: {e}") raise HTTPException(status_code=503, detail="External number service is currently unavailable.") # No need for a separate catch for HTTPException, it will propagate up # --- API Endpoints --- @app.on_event("startup") def on_startup(): init_database() @app.get("/", include_in_schema=False) def get_root(): return {"message": "API is running. Please go to /docs for documentation."} @app.get("/lookup/{phone_number}", response_model=NumberInfoResponse, tags=["Phone Lookup"]) def lookup_phone_number( phone_number: str = Path(..., description="The phone number to check (e.g., 923001234567).", regex="^\d+$") ): # <<< CRITICAL IMPROVEMENT #2 >>> # Normalize the phone number. Remove leading '+' or '00'. # For Pakistani numbers, a common mistake is entering 0300... which needs to be 92300... # This basic normalization helps, but a full library like 'phonenumbers' would be even better. if phone_number.startswith('0') and len(phone_number) > 10: # Simple heuristic for numbers like Pakistan's 03xx... # A more robust solution would be more complex. pass # For now, we assume the user enters the correct format e.g. 92300... cached_data = get_number_from_cache(phone_number) if cached_data: print(f"Cache HIT for {phone_number}") return NumberInfoResponse(**cached_data) print(f"Cache MISS for {phone_number}. Calling external service.") external_data = fetch_from_numverify(phone_number) # After the fetch, check if the number was actually valid according to the provider if not external_data.get('valid'): raise HTTPException( status_code=404, detail="The phone number is not valid or could not be found by the provider." ) save_number_to_cache(phone_number, external_data) response_data = { "phone_number": external_data.get('international_format'), "is_valid": external_data.get('valid'), "country_code": external_data.get('country_code'), "location": external_data.get('location'), "carrier": external_data.get('carrier'), "line_type": external_data.get('line_type') } return NumberInfoResponse(**response_data)