"""GenderAPI V2 teaching example: Python 3.10+, standard library, server-side only. Set GENDERAPI_API_KEY in your server environment. Importing this file makes no request. Run `python3 genderapi_v2.py --demo single` (or `--demo batch`) to spend credits. The 10-second timeout covers blocking socket operations, not the whole request. No redirects or retries: a lost response can still mean a completed, billed request. Reference: https://www.genderapi.io/docs/v2/responses """ from __future__ import annotations import argparse import http.client import json import os import re import sys import urllib.error import urllib.request BASE_URL = "https://api.genderapi.io/api/v2" TIMEOUT_SECONDS = 10 class GenderAPIError(Exception): """Keep the full parsed response for application handling; do not log it blindly.""" def __init__(self, message, *, status=None, response=None, headers=None): super().__init__(message) self.status = status self.response = response candidate = (response or {}).get("meta") meta = candidate if isinstance(candidate, dict) else {} references = (meta.get("request_id"), (response or {}).get("request_id"), (headers or {}).get("x-request-id")) self.request_id = next((value for value in references if isinstance(value, str) and value), None) self.retry_after = (headers or {}).get("retry-after") class _NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None # Do not forward the Bearer key to a redirect target. def make_item(input_type, value, *, ai_mode, country=None, item_id=None, force_to_genderize=False): """Choose off, fallback or always explicitly; country is optional ISO alpha-2.""" item = {"type": input_type, "value": value, "options": {"ai_mode": ai_mode}} if country is not None: item["country"] = country if item_id is not None: item["id"] = item_id if force_to_genderize: item["forceToGenderize"] = force_to_genderize if type(force_to_genderize) is not bool: raise ValueError("force_to_genderize must be a boolean") _validate_item(item) return item def _validate_item(item): if not isinstance(item, dict) or set(item) - {"type", "value", "country", "id", "options", "forceToGenderize"}: raise ValueError("Use make_item to build a supported prediction item") if item.get("type") not in ("name", "email", "username"): raise ValueError("type must be name, email or username") value = item.get("value") if not isinstance(value, str) or not value.strip() or len(value) > 254 or re.search(r"[\x00-\x1f\x7f]", value): raise ValueError("value must contain 1–254 characters without control characters") options = item.get("options") if not isinstance(options, dict) or set(options) != {"ai_mode"} or options["ai_mode"] not in ("off", "fallback", "always"): raise ValueError("Set options.ai_mode explicitly to off, fallback or always") force = item.get("forceToGenderize", False) if type(force) is not bool or (force and options["ai_mode"] != "fallback"): raise ValueError("forceToGenderize requires boolean true and ai_mode fallback") if "country" in item and (not isinstance(item["country"], str) or not re.fullmatch(r"[A-Z]{2}", item["country"])): raise ValueError("country must be an uppercase ISO alpha-2 code; omit unknown countries") if "id" in item and (not isinstance(item["id"], str) or not 1 <= len(item["id"]) <= 64): raise ValueError("id must contain 1–64 characters") # The API performs authoritative email and ISO-country membership validation. def _post(path, payload): key = os.environ.get("GENDERAPI_API_KEY", "") if not re.fullmatch(r"[a-fA-F0-9]{24}", key): raise ValueError("Set GENDERAPI_API_KEY to your 24-character hexadecimal API key in the server environment") request = urllib.request.Request( BASE_URL + path, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), headers={"Authorization": "Bearer " + key, "Content-Type": "application/json", "Accept": "application/json, application/problem+json"}, method="POST", ) try: try: response = urllib.request.build_opener(_NoRedirect()).open(request, timeout=TIMEOUT_SECONDS) except urllib.error.HTTPError as error: response = error # HTTPError bodies can still contain useful Problem JSON. with response: status, headers, raw = response.status, response.headers, response.read() except (OSError, urllib.error.URLError, http.client.HTTPException): raise GenderAPIError("Transport failed; billing outcome is unknown. Do not automatically retry.") from None if "json" not in headers.get("content-type", "").lower(): raise GenderAPIError("Expected JSON; inspect the request before another submission.", status=status, headers=headers) try: result = json.loads(raw) except (ValueError, UnicodeError): raise GenderAPIError("Invalid JSON; billing outcome is unknown.", status=status, headers=headers) from None if not isinstance(result, dict): raise GenderAPIError("Expected a JSON object.", status=status, headers=headers) if not 200 <= status < 300: raise GenderAPIError("API request failed; inspect response.code, response.action and response.meta.usage.", status=status, response=result, headers=headers) return result def _validate_success(result, items=None): def require(condition, message): if not condition: raise GenderAPIError(message, status=200, response=result) meta = result.get("meta") require(isinstance(meta, dict), "Missing response metadata.") require(isinstance(meta.get("request_id"), str) and bool(meta["request_id"]), "Missing response request ID.") access, usage = meta.get("access"), meta.get("usage") require(isinstance(access, dict) and access.get("mode") == "api_key", "Expected API-key access. Check your key; an IP-trial request may already have been charged.") require(isinstance(usage, dict) and usage.get("billing_status") == "confirmed", "Billing is not confirmed. Preserve the response and contact support before another submission.") require(type(usage.get("charged_credits")) is int and usage["charged_credits"] >= 0, "Invalid charged-credit metadata.") require(type(usage.get("remaining_credits")) is int, "Invalid remaining-credit metadata.") def prediction(data): require(isinstance(data, dict) and "gender" in data and "confidence" in data and "confidence_kind" in data, "Missing prediction fields.") require(data["gender"] in ("male", "female", None), "Unexpected gender result.") expected = "unknown" if data["gender"] is None else "identified" require(data.get("result_status") == expected, "Inconsistent prediction status.") if expected == "unknown": require(data["confidence"] is None and data["confidence_kind"] is None, "Unknown results must retain null confidence.") else: require(type(data["confidence"]) in (int, float) and 0 <= data["confidence"] <= 1 and data["confidence_kind"] in ("observed_frequency", "model_reported"), "Invalid confidence fields.") if items is None: prediction(result.get("data")) return result rows = result.get("data") require(isinstance(rows, list) and len(rows) == len(items), "Batch result count does not match the submitted items.") counts = {"total": len(items), "succeeded": 0, "identified": 0, "unknown": 0, "failed": 0} for index, row in enumerate(rows): require(isinstance(row, dict) and type(row.get("index")) is int and row["index"] == index and row.get("id") == items[index].get("id"), "Batch result mapping does not match the submitted items.") require(("data" in row) != ("error" in row), "Expected either item data or an item error.") require(type(row.get("charged_credits")) is int and row["charged_credits"] >= 0, "Missing item charge.") if "error" in row: require(isinstance(row["error"], dict) and isinstance(row["error"].get("code"), str) and row["charged_credits"] == 0, "Invalid failed-item outcome.") counts["failed"] += 1 else: prediction(row["data"]) counts["succeeded"] += 1 counts[row["data"]["result_status"]] += 1 require(isinstance(meta.get("summary"), dict) and all(meta["summary"].get(k) == v for k, v in counts.items()), "Batch summary does not match its item outcomes.") require(sum(row["charged_credits"] for row in rows) == usage["charged_credits"], "Batch charges do not match usage.") return result # Preserve unknowns, errors, all metadata and future additive fields. def predict(item): """One billable attempt; returns the complete response, including unknowns.""" _validate_item(item) return _validate_success(_post("/gender", item)) def predict_batch(items): """One 1–50 item batch; inspect every item even when HTTP status is 200.""" if not isinstance(items, list) or not 1 <= len(items) <= 50: raise ValueError("Provide a list of 1–50 items; chunk larger jobs yourself") for item in items: _validate_item(item) ids = [item["id"] for item in items if "id" in item] if len(ids) != len(set(ids)): raise ValueError("Batch IDs must be unique") return _validate_success(_post("/gender/batch", {"items": items}), items) def main(argv=None): parser = argparse.ArgumentParser(description="A server-side GenderAPI example. --demo makes a billable request.") parser.add_argument("--demo", choices=("single", "batch")) args = parser.parse_args(argv) if args.demo is None: parser.print_help() return 0 try: sample = make_item("name", "Alice Example", ai_mode="off", item_id="sample-1") result = predict(sample) if args.demo == "single" else predict_batch([sample]) # Print accounting and aggregate status only, not input values or the key. print(json.dumps({"result_status": result["data"].get("result_status") if args.demo == "single" else None, "summary": result["meta"].get("summary"), "usage": result["meta"]["usage"]}, indent=2)) return 2 if result["meta"].get("summary", {}).get("failed", 0) else 0 except (ValueError, GenderAPIError) as error: print(str(error), file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())