diff --git a/docs/modules.md b/docs/modules.md index 6454623f..9c9d064c 100644 --- a/docs/modules.md +++ b/docs/modules.md @@ -59,7 +59,7 @@ xml = XMLIngestor(validate_xsd="schema.xsd") sources = xml.ingest("data/records/") ``` -**Available ingestors:** `FileIngestor`, `WebIngestor`, `ParquetIngestor`, `XMLIngestor`, `RESTIngestor`, `DBIngestor`, `DuckDBIngestor`, `ElasticIngestor`, `EmailIngestor`, `FeedIngestor`, `GDriveIngestor`, `HuggingFaceIngestor`, `MCPIngestor`, `MongoIngestor`, `OntologyIngestor`, `PandasIngestor`, `RepoIngestor`, `SnowflakeIngestor`, `StreamIngestor` +**Available ingestors:** `FileIngestor`, `WebIngestor`, `ParquetIngestor`, `XMLIngestor`, `RESTIngestor`, `PublicAPIIngestor`, `DBIngestor`, `DuckDBIngestor`, `ElasticIngestor`, `EmailIngestor`, `FeedIngestor`, `GDriveIngestor`, `HuggingFaceIngestor`, `MCPIngestor`, `MongoIngestor`, `OntologyIngestor`, `PandasIngestor`, `RepoIngestor`, `SnowflakeIngestor`, `StreamIngestor` ### Parse diff --git a/docs/reference/ingest.md b/docs/reference/ingest.md index 036f3e79..114afcf7 100644 --- a/docs/reference/ingest.md +++ b/docs/reference/ingest.md @@ -1,6 +1,6 @@ --- title: "Ingest Module" -description: "Universal data ingestion from files, Parquet, XML, web, feeds, streams, repositories, email, and databases." +description: "Universal data ingestion from files, Parquet, XML, web, public APIs, feeds, streams, repositories, email, and databases." icon: "database" --- @@ -12,6 +12,8 @@ icon: "database" | --- | --- | | `FileIngestor` | PDF, DOCX, HTML, JSON, CSV, Excel, PPTX, ZIP/TAR — type auto-detected from extension | | `WebIngestor` | Web scraping and crawling with JavaScript rendering support | +| `RESTIngestor` | Generic REST API ingestion with headers, params, retries, pagination, and batch requests | +| `PublicAPIIngestor` | No-auth public API ingestion with examples, detection, rate limiting, and JSON/CSV/XML parsing | | `FeedIngestor` | RSS/Atom feed ingestion with live monitoring via `FeedMonitor` | | `StreamIngestor` | Real-time ingestion from Kafka, RabbitMQ, AWS Kinesis, and Apache Pulsar | | `RepoIngestor` | Git repositories — source files, commit history, README, and metadata | @@ -32,6 +34,9 @@ icon: "database" XXE-safe lxml with XSD/DTD validation and directory scanning (v0.5.0). + + Credential-free public API ingestion with pre-configured examples for JSONPlaceholder, REST Countries, Data.gov, and Open-Meteo. + Real-time ingestion from Kafka, RabbitMQ, AWS Kinesis, and Apache Pulsar. @@ -172,6 +177,44 @@ icon: "database" sources = ingestor.ingest_url("https://example.com/about") ``` + ### PublicAPIIngestor + + Use this for public REST-style APIs that do not require keys or tokens: + + ```python + from semantica.ingest import PublicAPIExamples, PublicAPIIngestor, ingest + + ingestor = PublicAPIIngestor(rate_limit_delay=1.0) + + posts = ingestor.ingest_public_api( + "https://jsonplaceholder.typicode.com/posts" + ) + + countries = ingestor.ingest_example("rest_countries_all") + + datasets = ingestor.ingest_example( + "data_gov_datasets", + params={"q": "transportation", "rows": 5}, + ) + + public = ingestor.detect_public_api( + "https://jsonplaceholder.typicode.com/posts" + ) + + result = ingest( + "https://jsonplaceholder.typicode.com/posts", + source_type="public_api", + ) + ``` + + Public API ingestion rejects common auth headers and query parameters by + default. Use `RESTIngestor` for authenticated APIs. + + ```python + examples = PublicAPIExamples.names() + mock_payload = PublicAPIExamples.sample_response("jsonplaceholder_posts") + ``` + ### FeedIngestor (RSS/Atom) ```python diff --git a/semantica/ingest/__init__.py b/semantica/ingest/__init__.py index 54238342..d35ce85d 100644 --- a/semantica/ingest/__init__.py +++ b/semantica/ingest/__init__.py @@ -76,7 +76,8 @@ Database Ingestion: - Multi-database Support: PostgreSQL, MySQL, SQLite, Oracle, SQL Server abstraction Key Features: - - Multiple ingestion source types (file, web, feed, stream, repo, email, db) + - Multiple ingestion source types (file, web, public API, feed, stream, + repo, email, db) - Unified ingestion function with source type dispatch - Method registry for extensibility - Configuration management with environment variables and config files @@ -87,6 +88,7 @@ Key Features: Main Classes: - FileIngestor: Local and cloud file processing - WebIngestor: Web scraping and crawling + - PublicAPIIngestor: No-auth public REST API processing - FeedIngestor: RSS/Atom feed processing - StreamIngestor: Real-time stream processing - RepoIngestor: Git repository processing @@ -102,6 +104,7 @@ Convenience Functions: - ingest: Unified ingestion function with source type dispatch - ingest_file: File ingestion wrapper - ingest_web: Web ingestion wrapper + - ingest_public_api: Public API ingestion wrapper - ingest_feed: Feed ingestion wrapper - ingest_stream: Stream ingestion wrapper - ingest_repository: Repository ingestion wrapper @@ -144,6 +147,7 @@ from .methods import ( ingest_mcp, ingest_ontology, ingest_parquet, + ingest_public_api, ingest_repository, ingest_stream, ingest_web, @@ -160,6 +164,13 @@ _LAZY_EXPORTS: Dict[str, Tuple[str, str]] = { "RobotsChecker": (".web_ingestor", "RobotsChecker"), "ContentExtractor": (".web_ingestor", "ContentExtractor"), "SitemapCrawler": (".web_ingestor", "SitemapCrawler"), + # REST and public API ingestion + "RESTIngestor": (".api_ingestor", "RESTIngestor"), + "APIData": (".api_ingestor", "APIData"), + "PublicAPIIngestor": (".public_api_ingestor", "PublicAPIIngestor"), + "PublicAPIExample": (".public_api_ingestor", "PublicAPIExample"), + "PublicAPIExamples": (".public_api_ingestor", "PublicAPIExamples"), + "PublicAPIDetection": (".public_api_ingestor", "PublicAPIDetection"), # Feed ingestion "FeedIngestor": (".feed_ingestor", "FeedIngestor"), "FeedItem": (".feed_ingestor", "FeedItem"), @@ -269,6 +280,13 @@ __all__ = [ "RobotsChecker", "ContentExtractor", "SitemapCrawler", + # REST and public API ingestion + "RESTIngestor", + "APIData", + "PublicAPIIngestor", + "PublicAPIExample", + "PublicAPIExamples", + "PublicAPIDetection", # Feed ingestion "FeedIngestor", "FeedItem", @@ -332,6 +350,7 @@ __all__ = [ "ingest_database", "ingest_ontology", "ingest_parquet", + "ingest_public_api", "ingest_xml", "ingest_mcp", "get_ingest_method", diff --git a/semantica/ingest/ingest_usage.md b/semantica/ingest/ingest_usage.md index 9fa60c38..78d2d36c 100644 --- a/semantica/ingest/ingest_usage.md +++ b/semantica/ingest/ingest_usage.md @@ -1,6 +1,6 @@ # Ingest Module Usage Guide -This guide demonstrates how to use the ingest module for ingesting data from various sources including files, XML, web content, feeds, streams, repositories, emails, and databases. +This guide demonstrates how to use the ingest module for ingesting data from various sources including files, XML, web content, public APIs, feeds, streams, repositories, emails, and databases. ## Table of Contents @@ -9,17 +9,18 @@ This guide demonstrates how to use the ingest module for ingesting data from var 3. [Parquet Ingestion](#parquet-ingestion) 4. [XML Ingestion](#xml-ingestion) 5. [Web Ingestion](#web-ingestion) -6. [Feed Ingestion](#feed-ingestion) -7. [Stream Ingestion](#stream-ingestion) -8. [Repository Ingestion](#repository-ingestion) -9. [Email Ingestion](#email-ingestion) -10. [Database Ingestion](#database-ingestion) -11. [MCP Server Ingestion](#mcp-server-ingestion) -12. [Unified Ingestion](#unified-ingestion) -13. [Using Methods](#using-methods) -14. [Using Registry](#using-registry) -15. [Configuration](#configuration) -16. [Advanced Examples](#advanced-examples) +6. [Public API Ingestion](#public-api-ingestion) +7. [Feed Ingestion](#feed-ingestion) +8. [Stream Ingestion](#stream-ingestion) +9. [Repository Ingestion](#repository-ingestion) +10. [Email Ingestion](#email-ingestion) +11. [Database Ingestion](#database-ingestion) +12. [MCP Server Ingestion](#mcp-server-ingestion) +13. [Unified Ingestion](#unified-ingestion) +14. [Using Methods](#using-methods) +15. [Using Registry](#using-registry) +16. [Configuration](#configuration) +17. [Advanced Examples](#advanced-examples) ## Basic Usage @@ -40,6 +41,12 @@ result = ingest("catalog.xml") # Ingest from web URL result = ingest("https://example.com", source_type="web") +# Ingest from a public API with no authentication +result = ingest( + "https://jsonplaceholder.typicode.com/posts", + source_type="public_api", +) + # Ingest from feed result = ingest("https://example.com/feed.xml", source_type="feed") ``` @@ -47,12 +54,13 @@ result = ingest("https://example.com/feed.xml", source_type="feed") ### Using Main Classes ```python -from semantica.ingest import FileIngestor, WebIngestor, XMLIngestor +from semantica.ingest import FileIngestor, PublicAPIIngestor, WebIngestor, XMLIngestor # Create ingestor file_ingestor = FileIngestor() web_ingestor = WebIngestor(delay=1.0, respect_robots=True) xml_ingestor = XMLIngestor() +api_ingestor = PublicAPIIngestor(rate_limit_delay=1.0) # Ingest files files = file_ingestor.ingest_directory("./documents", recursive=True) @@ -62,6 +70,11 @@ content = web_ingestor.ingest_url("https://example.com") # Ingest XML content xml_data = xml_ingestor.ingest_file("catalog.xml") + +# Ingest public API records without credentials +api_data = api_ingestor.ingest_public_api( + "https://jsonplaceholder.typicode.com/posts" +) ``` ## File Ingestion @@ -356,6 +369,117 @@ metadata = extractor.extract_metadata(html, url="https://example.com") links = extractor.extract_links(html, base_url="https://example.com") ``` +## Public API Ingestion + +Public API ingestion is for REST-style endpoints that do not require API keys, +OAuth tokens, or other credentials. It is useful for examples, tests, CI, and +contributor development because all requests are made without authentication. + +Use `RESTIngestor` instead when an endpoint requires `Authorization`, +`X-API-Key`, `api_key`, or similar credentials. + +### Public Endpoint Ingestion + +```python +from semantica.ingest import PublicAPIIngestor, ingest_public_api + +# Using convenience function +posts = ingest_public_api( + "https://jsonplaceholder.typicode.com/posts", + rate_limit_delay=1.0, +) + +# Using class directly +ingestor = PublicAPIIngestor(rate_limit_delay=1.0) +users = ingestor.ingest_public_api( + "https://jsonplaceholder.typicode.com/users", +) + +print(posts.metadata["record_count"]) +print(users.data[0]["name"]) +``` + +### Pre-Configured Public API Examples + +```python +from semantica.ingest import PublicAPIExamples, PublicAPIIngestor + +print(PublicAPIExamples.names()) +print(PublicAPIExamples.endpoints()) + +ingestor = PublicAPIIngestor(rate_limit_delay=1.0) + +# JSONPlaceholder: fake REST resources for testing +posts = ingestor.ingest_example("jsonplaceholder_posts") + +# REST Countries: country reference data +countries = ingestor.ingest_example("rest_countries_all") + +# Data.gov catalog search: nested records extracted from result.results +datasets = ingestor.ingest_example( + "data_gov_datasets", + params={"q": "transportation", "rows": 5}, +) +``` + +### Public API Detection + +```python +from semantica.ingest import PublicAPIIngestor + +ingestor = PublicAPIIngestor(rate_limit_delay=1.0) + +detection = ingestor.detect_public_api( + "https://jsonplaceholder.typicode.com/posts" +) + +print(detection.is_public) +print(detection.requires_auth) +print(detection.response_status) +``` + +Detection is endpoint-level. A successful no-auth response means that endpoint +appears public; it does not prove that every endpoint on the same API is public. + +### JSON, CSV, and XML Responses + +```python +from semantica.ingest import PublicAPIIngestor + +ingestor = PublicAPIIngestor(rate_limit_delay=1.0) + +# JSON is auto-detected from Content-Type or response body +json_data = ingestor.ingest_public_api( + "https://jsonplaceholder.typicode.com/todos" +) + +# CSV can be parsed into dictionaries +csv_data = ingestor.ingest_public_api( + "https://example.com/data.csv", + response_format="csv", +) + +# XML is converted to nested dictionaries; record_path can select child nodes +xml_data = ingestor.ingest_public_api( + "https://example.com/data.xml", + response_format="xml", + record_path="children", +) +``` + +### Testing Helpers + +```python +from semantica.ingest import PublicAPIExamples + +# Mock payloads for unit tests without live network calls +payload = PublicAPIExamples.sample_response("jsonplaceholder_posts") +data_gov_payload = PublicAPIExamples.sample_response("data_gov_datasets") +``` + +These fixtures are intentionally small and credential-free. Use mocked HTTP +responses in CI rather than depending on public API uptime. + ## Feed Ingestion ### RSS Feed Ingestion @@ -1076,6 +1200,15 @@ result = ingest("postgresql://user:pass@localhost/db") # Auto-detects database result = ingest("http://localhost:8000/mcp", source_type="mcp") # MCP server ingestion via URL ``` +Public APIs should be explicit because regular URLs default to web ingestion: + +```python +result = ingest( + "https://jsonplaceholder.typicode.com/posts", + source_type="public_api", +) +``` + ### Explicit Source Type ```python @@ -1085,6 +1218,7 @@ from semantica.ingest import ingest result = ingest("document.pdf", source_type="file") result = ingest("catalog.xml", source_type="xml") result = ingest("https://example.com", source_type="web") +result = ingest("https://jsonplaceholder.typicode.com/posts", source_type="public_api") result = ingest("https://example.com/feed.xml", source_type="feed") ``` @@ -1112,6 +1246,7 @@ print(f"Ingested {len(result['files'])} files") from semantica.ingest.methods import ( ingest_file, ingest_web, + ingest_public_api, ingest_feed, ingest_stream, ingest_repository, @@ -1128,6 +1263,12 @@ files = ingest_file("./documents", method="directory") # Web ingestion content = ingest_web("https://example.com", method="url") +# Public API ingestion +api_data = ingest_public_api( + "https://jsonplaceholder.typicode.com/posts", + method="endpoint", +) + # Feed ingestion feed = ingest_feed("https://example.com/feed.xml", method="rss") diff --git a/semantica/ingest/methods.py b/semantica/ingest/methods.py index 9d56a41b..52572365 100644 --- a/semantica/ingest/methods.py +++ b/semantica/ingest/methods.py @@ -30,6 +30,12 @@ Web Ingestion: - "sitemap": Sitemap-based crawling - "crawl": Domain crawling +Public API Ingestion: + - "endpoint": No-auth public API endpoint ingestion + - "example": Pre-configured public API examples + - "detect": Endpoint-level public/no-auth detection + - "batch": Multiple no-auth public API endpoints + Feed Ingestion: - "rss": RSS feed ingestion - "atom": Atom feed ingestion @@ -138,6 +144,7 @@ Key Features: Main Functions: - ingest_file: File ingestion wrapper - ingest_web: Web ingestion wrapper + - ingest_public_api: No-auth public API ingestion wrapper - ingest_feed: Feed ingestion wrapper - ingest_stream: Stream ingestion wrapper - ingest_repository: Repository ingestion wrapper @@ -171,12 +178,14 @@ from .file_ingestor import FileIngestor, FileObject from .registry import method_registry if TYPE_CHECKING: + from .api_ingestor import APIData from .db_ingestor import TableData from .email_ingestor import EmailData from .feed_ingestor import FeedData from .mcp_ingestor import MCPData from .ontology_ingestor import OntologyData from .parquet_ingestor import ParquetData + from .public_api_ingestor import PublicAPIDetection from .stream_ingestor import StreamProcessor from .web_ingestor import WebContent from .xml_ingestor import XMLIngestionData @@ -483,6 +492,105 @@ def ingest_web( raise +def ingest_public_api( + source: Union[str, List[str]], + method: str = "endpoint", + **kwargs, +) -> Union[ + APIData, + List[APIData], + PublicAPIDetection, + List[PublicAPIDetection], + Dict[str, Any], +]: + """ + Ingest public, no-auth API endpoints. + + Args: + source: Endpoint URL, example name, or list of endpoint URLs/example names + method: Public API ingestion method: + - "endpoint": Ingest a single no-auth endpoint + - "example": Ingest a pre-configured public API example + - "detect": Check whether endpoint access works without auth + - "batch": Ingest multiple no-auth endpoints + - "examples": List pre-configured public API examples + **kwargs: Additional options passed to PublicAPIIngestor. Use + ``http_method`` to override the HTTP method because ``method`` is + reserved for ingestion dispatch. + + Returns: + APIData, list of APIData, detection result(s), or examples dictionary + + Examples: + >>> from semantica.ingest.methods import ingest_public_api + >>> data = ingest_public_api("https://jsonplaceholder.typicode.com/posts") + >>> countries = ingest_public_api("rest_countries_all", method="example") + >>> detection = ingest_public_api( + ... "https://jsonplaceholder.typicode.com/posts", + ... method="detect", + ... ) + """ + custom_method = method_registry.get("public_api", method) + if custom_method and custom_method != ingest_public_api: + try: + return custom_method(source, **kwargs) + except Exception as e: + logger.warning( + f"Custom method {method} failed: {e}, falling back to default" + ) + + try: + from .public_api_ingestor import PublicAPIExamples, PublicAPIIngestor + + config = ingest_config.get_method_config("public_api") + config.update(kwargs) + ingestor = PublicAPIIngestor(**config) + + request_kwargs = kwargs.copy() + for config_only_key in ( + "backoff_factor", + "delay", + "fail_fast", + "max_retries", + "validate_no_auth", + ): + request_kwargs.pop(config_only_key, None) + + http_method = request_kwargs.pop("http_method", None) + if http_method: + request_kwargs["method"] = http_method + + if method in {"examples", "list_examples"}: + tag = request_kwargs.pop("tag", None) + return {"examples": PublicAPIExamples.list_examples(tag=tag)} + + if method in {"detect", "detection"}: + if isinstance(source, list): + return [ + ingestor.detect_public_api(endpoint, **request_kwargs) + for endpoint in source + ] + return ingestor.detect_public_api(source, **request_kwargs) + + if method in {"example", "sample"}: + if isinstance(source, list): + return [ + ingestor.ingest_example(name, **request_kwargs) for name in source + ] + return ingestor.ingest_example(source, **request_kwargs) + + if method == "batch" or isinstance(source, list): + if not isinstance(source, list): + raise ProcessingError("Public API batch ingestion requires a list") + return ingestor.batch_public_apis(source, **request_kwargs) + + return ingestor.ingest_public_api(source, **request_kwargs) + + except Exception as e: + logger.error(f"Failed to ingest public API: {e}") + raise + + def ingest_feed( source: Union[str, List[str]], method: str = "rss", **kwargs ) -> Union[FeedData, List[FeedData], Dict[str, Any]]: @@ -1085,6 +1193,7 @@ def ingest( source_type: Source type (auto-detected if not specified) - "file": File ingestion - "web": Web ingestion + - "public_api": No-auth public API ingestion - "feed": Feed ingestion - "stream": Stream ingestion - "repo": Repository ingestion @@ -1100,9 +1209,9 @@ def ingest( Dict with ingestion results. The top-level key depends on source_type: - "files": file ingestion - "content": web ingestion + - "data": public API, database, parquet, or MCP ingestion - "feeds": feed ingestion - "emails": email ingestion - - "data": database, parquet, or MCP ingestion - "ontology": ontology ingestion - "xml": XML file or directory ingestion (use ``result["xml"]``) @@ -1172,6 +1281,10 @@ def ingest( return {"files": ingest_file(sources, method=method or "file", **kwargs)} elif source_type == "web": return {"content": ingest_web(sources, method=method or "url", **kwargs)} + elif source_type in {"public_api", "api"}: + return { + "data": ingest_public_api(sources, method=method or "endpoint", **kwargs) + } elif source_type == "feed": return {"feeds": ingest_feed(sources, method=method or "rss", **kwargs)} elif source_type == "stream": @@ -1250,6 +1363,17 @@ method_registry.register("web", "default", ingest_web) method_registry.register("web", "url", ingest_web) method_registry.register("web", "sitemap", ingest_web) method_registry.register("web", "crawl", ingest_web) +method_registry.register("public_api", "default", ingest_public_api) +method_registry.register("public_api", "endpoint", ingest_public_api) +method_registry.register("public_api", "example", ingest_public_api) +method_registry.register("public_api", "sample", ingest_public_api) +method_registry.register("public_api", "detect", ingest_public_api) +method_registry.register("public_api", "detection", ingest_public_api) +method_registry.register("public_api", "batch", ingest_public_api) +method_registry.register("public_api", "examples", ingest_public_api) +method_registry.register("public_api", "list_examples", ingest_public_api) +method_registry.register("api", "public", ingest_public_api) +method_registry.register("api", "endpoint", ingest_public_api) method_registry.register("feed", "default", ingest_feed) method_registry.register("feed", "rss", ingest_feed) method_registry.register("feed", "atom", ingest_feed) diff --git a/semantica/ingest/public_api_ingestor.py b/semantica/ingest/public_api_ingestor.py new file mode 100644 index 00000000..b13aa0de --- /dev/null +++ b/semantica/ingest/public_api_ingestor.py @@ -0,0 +1,794 @@ +""" +Public API Ingestion Module + +This module provides no-auth public API ingestion built on top of the generic +RESTIngestor. It focuses on contributor-friendly endpoints, public API +detection, polite rate limiting, and response normalization for JSON, CSV, and +XML APIs. + +Example Usage: + >>> from semantica.ingest import PublicAPIIngestor + >>> ingestor = PublicAPIIngestor(rate_limit_delay=1.0) + >>> data = ingestor.ingest_public_api( + ... "https://jsonplaceholder.typicode.com/posts" + ... ) + >>> data.metadata["record_count"] + 100 +""" + +from __future__ import annotations + +import copy +import csv +import io +import time +import xml.etree.ElementTree as ET +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Dict, List, Optional, Tuple +from urllib.parse import urlparse + +import requests + +from ..utils.exceptions import ProcessingError, ValidationError +from ..utils.logging import get_logger +from .api_ingestor import APIData, RESTIngestor + +AUTH_HEADER_NAMES = { + "authorization", + "proxy-authorization", + "x-api-key", + "api-key", + "apikey", + "ocp-apim-subscription-key", + "x-rapidapi-key", + "x-auth-token", + "x-access-token", +} + +AUTH_PARAM_NAMES = { + "api_key", + "apikey", + "access_token", + "auth_token", + "bearer_token", + "client_secret", + "token", + "subscription_key", + "subscription-key", +} + + +@dataclass +class PublicAPIExample: + """Pre-configured public API endpoint definition.""" + + name: str + endpoint: str + description: str + method: str = "GET" + params: Dict[str, Any] = field(default_factory=dict) + headers: Dict[str, str] = field(default_factory=dict) + response_format: str = "auto" + record_path: Optional[str] = None + tags: List[str] = field(default_factory=list) + rate_limit_delay: Optional[float] = None + + +@dataclass +class PublicAPIDetection: + """Result of checking whether an endpoint can be reached without auth.""" + + endpoint: str + is_public: bool + requires_auth: bool + response_status: Optional[int] = None + reason: str = "" + metadata: Dict[str, Any] = field(default_factory=dict) + checked_at: datetime = field(default_factory=datetime.now) + + +class PublicAPIExamples: + """ + Catalog of no-auth public API examples for tests and demos. + + The examples intentionally avoid credentials and use endpoints suitable for + CI-friendly mocked tests or quick local experiments. + """ + + _EXAMPLES: Dict[str, PublicAPIExample] = { + "jsonplaceholder_posts": PublicAPIExample( + name="jsonplaceholder_posts", + endpoint="https://jsonplaceholder.typicode.com/posts", + description="Fake blog post records for REST API testing.", + response_format="json", + tags=["json", "testing", "placeholder"], + rate_limit_delay=0.5, + ), + "jsonplaceholder_users": PublicAPIExample( + name="jsonplaceholder_users", + endpoint="https://jsonplaceholder.typicode.com/users", + description="Fake user records for REST API testing.", + response_format="json", + tags=["json", "testing", "placeholder"], + rate_limit_delay=0.5, + ), + "jsonplaceholder_todos": PublicAPIExample( + name="jsonplaceholder_todos", + endpoint="https://jsonplaceholder.typicode.com/todos", + description="Fake todo records for REST API testing.", + response_format="json", + tags=["json", "testing", "placeholder"], + rate_limit_delay=0.5, + ), + "rest_countries_all": PublicAPIExample( + name="rest_countries_all", + endpoint="https://restcountries.com/v3.1/all", + description="Country reference data from REST Countries.", + params={ + "fields": "name,capital,region,population,cca2,cca3", + }, + response_format="json", + tags=["json", "countries", "reference"], + rate_limit_delay=1.0, + ), + "data_gov_datasets": PublicAPIExample( + name="data_gov_datasets", + endpoint="https://catalog.data.gov/api/3/action/package_search", + description="Data.gov catalog package search results.", + params={"q": "climate", "rows": 10}, + response_format="json", + record_path="result.results", + tags=["json", "government", "datasets"], + rate_limit_delay=1.0, + ), + "open_meteo_forecast": PublicAPIExample( + name="open_meteo_forecast", + endpoint="https://api.open-meteo.com/v1/forecast", + description="Open-Meteo forecast sample for Berlin.", + params={ + "latitude": 52.52, + "longitude": 13.41, + "current": "temperature_2m,wind_speed_10m", + }, + response_format="json", + tags=["json", "weather", "forecast"], + rate_limit_delay=1.0, + ), + } + + _SAMPLE_RESPONSES: Dict[str, Any] = { + "jsonplaceholder_posts": [ + {"userId": 1, "id": 1, "title": "sample post", "body": "body text"} + ], + "jsonplaceholder_users": [ + {"id": 1, "name": "Leanne Graham", "email": "leanne@example.com"} + ], + "jsonplaceholder_todos": [ + {"userId": 1, "id": 1, "title": "sample todo", "completed": False} + ], + "rest_countries_all": [ + { + "name": {"common": "India", "official": "Republic of India"}, + "capital": ["New Delhi"], + "region": "Asia", + "population": 1407563842, + "cca2": "IN", + "cca3": "IND", + } + ], + "data_gov_datasets": { + "success": True, + "result": { + "count": 1, + "results": [ + { + "id": "sample-dataset", + "title": "Sample Dataset", + "metadata_created": "2026-01-01T00:00:00", + } + ], + }, + }, + "open_meteo_forecast": { + "latitude": 52.52, + "longitude": 13.41, + "current": {"temperature_2m": 18.2, "wind_speed_10m": 9.1}, + }, + } + + @classmethod + def list_examples(cls, tag: Optional[str] = None) -> List[PublicAPIExample]: + """ + List available public API examples. + + Args: + tag: Optional tag filter such as "json", "government", or "testing" + + Returns: + List of public API example definitions + """ + examples = cls._EXAMPLES.values() + if tag: + tag_lower = tag.lower() + examples = [example for example in examples if tag_lower in example.tags] + return [copy.deepcopy(example) for example in examples] + + @classmethod + def names(cls, tag: Optional[str] = None) -> List[str]: + """Return example names, optionally filtered by tag.""" + return [example.name for example in cls.list_examples(tag=tag)] + + @classmethod + def endpoints(cls) -> Dict[str, str]: + """Return a mapping of example name to endpoint URL.""" + return {name: example.endpoint for name, example in cls._EXAMPLES.items()} + + @classmethod + def get(cls, name: str) -> PublicAPIExample: + """ + Get a public API example by name. + + Args: + name: Example name. Hyphens are normalized to underscores. + + Raises: + ValidationError: If the example is unknown + """ + normalized_name = name.lower().replace("-", "_") + if normalized_name not in cls._EXAMPLES: + available = ", ".join(sorted(cls._EXAMPLES)) + raise ValidationError( + f"Unknown public API example: {name}. Available examples: {available}" + ) + return copy.deepcopy(cls._EXAMPLES[normalized_name]) + + @classmethod + def sample_response(cls, name: str) -> Any: + """ + Return a small mock response payload for an example. + + These fixtures are intended for tests and documentation snippets so + contributors can exercise ingestion without making live network calls. + """ + normalized_name = name.lower().replace("-", "_") + if normalized_name not in cls._SAMPLE_RESPONSES: + available = ", ".join(sorted(cls._SAMPLE_RESPONSES)) + raise ValidationError( + f"No sample response for public API example: {name}. " + f"Available samples: {available}" + ) + return copy.deepcopy(cls._SAMPLE_RESPONSES[normalized_name]) + + +class PublicAPIIngestor(RESTIngestor): + """ + Public, no-auth API ingestion handler. + + PublicAPIIngestor uses RESTIngestor's HTTP session and retry behavior while + adding no-auth validation, endpoint-level public detection, response + normalization, and built-in public API examples. + """ + + def __init__( + self, + config: Optional[Dict[str, Any]] = None, + rate_limit_delay: Optional[float] = None, + validate_no_auth: Optional[bool] = None, + **kwargs, + ): + """ + Initialize public API ingestor. + + Args: + config: Optional ingestion configuration dictionary + rate_limit_delay: Minimum seconds between public API requests + validate_no_auth: Reject auth headers/params before requests + **kwargs: Additional configuration values + """ + merged_config = (config or {}).copy() + merged_config.update(kwargs) + if rate_limit_delay is not None: + merged_config["rate_limit_delay"] = rate_limit_delay + if validate_no_auth is not None: + merged_config["validate_no_auth"] = validate_no_auth + + super().__init__(config=merged_config) + self.logger = get_logger("public_api_ingestor") + self.rate_limit_delay = float( + self.config.get("rate_limit_delay", self.config.get("delay", 1.0)) or 0.0 + ) + self.validate_no_auth = bool(self.config.get("validate_no_auth", True)) + self._last_request_time = 0.0 + + self.logger.debug("Public API ingestor initialized") + + def detect_public_api( + self, + endpoint: str, + method: str = "GET", + headers: Optional[Dict[str, str]] = None, + params: Optional[Dict[str, Any]] = None, + **options, + ) -> PublicAPIDetection: + """ + Detect whether an endpoint is reachable without authentication. + + Detection is endpoint-level: a successful unauthenticated request means + this specific endpoint appears public, not necessarily the entire API. + """ + self._validate_endpoint(endpoint) + auth_indicators = self._auth_indicators( + headers=headers, params=params, options=options + ) + if auth_indicators: + return PublicAPIDetection( + endpoint=endpoint, + is_public=False, + requires_auth=True, + reason=( + "Authentication credentials were provided; " + "no-auth access was not tested." + ), + metadata={"auth_indicators": auth_indicators}, + ) + + request_options = options.copy() + timeout = request_options.pop("timeout", self.config.get("timeout", 30)) + rate_limit_delay = request_options.pop("rate_limit_delay", None) + request_headers = self._merged_headers(headers) + + try: + self._wait_if_needed(rate_limit_delay=rate_limit_delay) + response = self.session.request( + method=method, + url=endpoint, + headers=request_headers, + params=params, + timeout=timeout, + **request_options, + ) + except requests.exceptions.RequestException as exc: + self.logger.error(f"Failed to detect public API {endpoint}: {exc}") + raise ProcessingError(f"Failed to detect public API: {exc}") from exc + + is_public, requires_auth, reason = self._classify_public_response(response) + return PublicAPIDetection( + endpoint=endpoint, + is_public=is_public, + requires_auth=requires_auth, + response_status=response.status_code, + reason=reason, + metadata={ + "method": method, + "content_type": response.headers.get("Content-Type"), + "www_authenticate": response.headers.get("WWW-Authenticate"), + }, + ) + + def is_public_api(self, endpoint: str, **options) -> bool: + """Return True when an endpoint appears reachable without auth.""" + return self.detect_public_api(endpoint, **options).is_public + + def ingest_public_api( + self, + endpoint: str, + method: str = "GET", + headers: Optional[Dict[str, str]] = None, + params: Optional[Dict[str, Any]] = None, + data: Optional[Any] = None, + json_data: Optional[Dict[str, Any]] = None, + response_format: str = "auto", + record_path: Optional[str] = None, + normalize_records: bool = True, + require_public: bool = True, + rate_limit_delay: Optional[float] = None, + **options, + ) -> APIData: + """ + Ingest data from a public API endpoint without authentication. + + Args: + endpoint: Public API endpoint URL + method: HTTP method, usually GET + headers: Optional non-auth request headers + params: Optional query parameters + data: Optional request body + json_data: Optional JSON request body + response_format: "auto", "json", "csv", "xml", or "text" + record_path: Dot path to records inside nested JSON/XML data + normalize_records: Convert response into a list of dictionaries + require_public: Raise if response indicates auth is required + rate_limit_delay: Per-request delay override + **options: Additional requests options + + Returns: + APIData: Normalized public API response and metadata + """ + self._validate_endpoint(endpoint) + self._validate_no_auth_request(headers=headers, params=params, options=options) + + tracking_id = self.progress_tracker.start_tracking( + file=endpoint, + module="ingest", + submodule="PublicAPIIngestor", + message=f"Requesting public API: {method} {endpoint}", + ) + + request_options = options.copy() + timeout = request_options.pop("timeout", self.config.get("timeout", 30)) + request_headers = self._merged_headers(headers) + + try: + self._wait_if_needed(rate_limit_delay=rate_limit_delay) + response = self.session.request( + method=method, + url=endpoint, + headers=request_headers, + params=params, + data=data, + json=json_data, + timeout=timeout, + **request_options, + ) + + is_public, requires_auth, reason = self._classify_public_response(response) + if require_public and requires_auth: + raise ValidationError( + f"Endpoint appears to require authentication: {endpoint} " + f"({response.status_code}). Use RESTIngestor for " + "authenticated APIs." + ) + + response.raise_for_status() + parsed_data, detected_format = self._parse_response( + response=response, + response_format=response_format, + endpoint=endpoint, + ) + + if normalize_records: + result_data = self._to_records(parsed_data, record_path=record_path) + elif record_path: + result_data = self._extract_record_path(parsed_data, record_path) + else: + result_data = parsed_data + + record_count = len(result_data) if isinstance(result_data, list) else 1 + + self.progress_tracker.stop_tracking( + tracking_id, + status="completed", + message=f"Public API request successful: {response.status_code}", + ) + self.logger.info( + f"Public API request completed: {method} {endpoint} - " + f"{response.status_code}" + ) + + return APIData( + data=result_data, + response_status=response.status_code, + endpoint=endpoint, + metadata={ + "method": method, + "headers": dict(response.headers), + "content_type": response.headers.get("Content-Type"), + "public_api": is_public, + "authentication": "none", + "public_detection_reason": reason, + "requires_auth": requires_auth, + "response_format": detected_format, + "record_path": record_path, + "normalized_records": normalize_records, + "record_count": record_count, + "source_data_type": type(parsed_data).__name__, + }, + ) + + except (ValidationError, ProcessingError): + self.progress_tracker.stop_tracking( + tracking_id, status="failed", message="Public API ingestion failed" + ) + raise + except requests.exceptions.RequestException as exc: + self.progress_tracker.stop_tracking( + tracking_id, status="failed", message=str(exc) + ) + self.logger.error(f"Failed to ingest public API {endpoint}: {exc}") + raise ProcessingError(f"Failed to ingest public API: {exc}") from exc + except Exception as exc: + self.progress_tracker.stop_tracking( + tracking_id, status="failed", message=str(exc) + ) + self.logger.error(f"Failed to ingest public API {endpoint}: {exc}") + raise ProcessingError(f"Failed to ingest public API: {exc}") from exc + + def ingest_example(self, name: str, **overrides) -> APIData: + """ + Ingest one of the pre-configured public API examples. + + Args: + name: Example name from PublicAPIExamples + **overrides: Optional endpoint, params, headers, record_path, or + response_format overrides + """ + example = PublicAPIExamples.get(name) + params = example.params.copy() + params.update(overrides.pop("params", {}) or {}) + headers = example.headers.copy() + headers.update(overrides.pop("headers", {}) or {}) + + endpoint = overrides.pop("endpoint", example.endpoint) + method = overrides.pop("method", example.method) + response_format = overrides.pop("response_format", example.response_format) + record_path = overrides.pop("record_path", example.record_path) + rate_limit_delay = overrides.pop("rate_limit_delay", example.rate_limit_delay) + + result = self.ingest_public_api( + endpoint=endpoint, + method=method, + headers=headers or None, + params=params or None, + response_format=response_format, + record_path=record_path, + rate_limit_delay=rate_limit_delay, + **overrides, + ) + result.metadata["example_name"] = example.name + result.metadata["example_description"] = example.description + result.metadata["example_tags"] = example.tags + return result + + def ingest_examples(self, names: List[str], **options) -> List[APIData]: + """Ingest multiple pre-configured public API examples.""" + return [self.ingest_example(name, **options) for name in names] + + def batch_public_apis( + self, + endpoints: List[str], + method: str = "GET", + **options, + ) -> List[APIData]: + """Ingest multiple public API endpoints with no-auth validation.""" + results: List[APIData] = [] + for endpoint in endpoints: + try: + results.append( + self.ingest_public_api(endpoint, method=method, **options) + ) + except Exception as exc: + self.logger.warning(f"Failed to fetch public API {endpoint}: {exc}") + if self.config.get("fail_fast", False): + raise + return results + + def _validate_endpoint(self, endpoint: str) -> None: + parsed = urlparse(endpoint) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValidationError( + f"Public API endpoint must be an absolute HTTP(S) URL: {endpoint}" + ) + + def _merged_headers( + self, headers: Optional[Dict[str, str]] = None + ) -> Dict[str, str]: + base_headers = dict(getattr(self.session, "headers", {}) or {}) + if headers: + base_headers.update(headers) + return base_headers + + def _auth_indicators( + self, + headers: Optional[Dict[str, str]] = None, + params: Optional[Dict[str, Any]] = None, + options: Optional[Dict[str, Any]] = None, + ) -> List[str]: + indicators: List[str] = [] + merged_headers = self._merged_headers(headers) + for header_name in merged_headers: + if header_name.lower() in AUTH_HEADER_NAMES: + indicators.append(f"header:{header_name}") + + for param_name in params or {}: + if param_name.lower() in AUTH_PARAM_NAMES: + indicators.append(f"param:{param_name}") + + if options and options.get("auth") is not None: + indicators.append("request:auth") + + return indicators + + def _validate_no_auth_request( + self, + headers: Optional[Dict[str, str]] = None, + params: Optional[Dict[str, Any]] = None, + options: Optional[Dict[str, Any]] = None, + ) -> None: + if not self.validate_no_auth: + return + + auth_indicators = self._auth_indicators( + headers=headers, + params=params, + options=options, + ) + if auth_indicators: + indicators = ", ".join(auth_indicators) + raise ValidationError( + "Public API ingestion only supports no-auth endpoints. " + f"Authentication indicators found: {indicators}. " + "Use RESTIngestor for authenticated APIs." + ) + + def _wait_if_needed(self, rate_limit_delay: Optional[float] = None) -> None: + delay = self.rate_limit_delay if rate_limit_delay is None else rate_limit_delay + delay = float(delay or 0.0) + now = time.time() + + if delay > 0 and self._last_request_time: + elapsed = now - self._last_request_time + if elapsed < delay: + time.sleep(delay - elapsed) + + self._last_request_time = time.time() + + def _classify_public_response( + self, response: requests.Response + ) -> Tuple[bool, bool, str]: + status_code = response.status_code + if 200 <= status_code < 300: + return True, False, "Endpoint responded successfully without auth." + if status_code in {401, 403}: + return ( + False, + True, + "Endpoint returned an authentication or authorization status.", + ) + if status_code == 429: + return False, False, "Endpoint is rate limited; public access is unclear." + return False, False, f"Endpoint returned status {status_code}." + + def _parse_response( + self, + response: requests.Response, + response_format: str, + endpoint: str, + ) -> Tuple[Any, str]: + detected_format = self._detect_response_format( + response=response, + requested_format=response_format, + endpoint=endpoint, + ) + + try: + if detected_format == "json": + return response.json(), "json" + if detected_format == "csv": + return self._parse_csv(response.text), "csv" + if detected_format == "xml": + return self._parse_xml(response.text), "xml" + return response.text, "text" + except ValueError as exc: + raise ProcessingError( + f"Failed to parse {detected_format.upper()} public API response" + ) from exc + except ET.ParseError as exc: + raise ProcessingError("Failed to parse XML public API response") from exc + + def _detect_response_format( + self, + response: requests.Response, + requested_format: str, + endpoint: str, + ) -> str: + requested = requested_format.lower() + if requested != "auto": + if requested not in {"json", "csv", "xml", "text"}: + raise ValidationError( + "response_format must be one of: auto, json, csv, xml, text" + ) + return requested + + content_type = response.headers.get("Content-Type", "").lower() + endpoint_lower = endpoint.lower() + text = (response.text or "").lstrip() + + if "json" in content_type or endpoint_lower.endswith(".json"): + return "json" + if "csv" in content_type or endpoint_lower.endswith(".csv"): + return "csv" + if "xml" in content_type or endpoint_lower.endswith(".xml"): + return "xml" + if text.startswith(("{", "[")): + return "json" + if text.startswith("<"): + return "xml" + return "text" + + def _parse_csv(self, csv_text: str) -> List[Dict[str, Any]]: + reader = csv.DictReader(io.StringIO(csv_text)) + return [dict(row) for row in reader] + + def _parse_xml(self, xml_text: str) -> Dict[str, Any]: + root = ET.fromstring(xml_text) + return self._element_to_dict(root) + + def _element_to_dict(self, element: ET.Element) -> Dict[str, Any]: + children = [self._element_to_dict(child) for child in list(element)] + return { + "tag": self._strip_namespace(element.tag), + "attributes": { + self._strip_namespace(key): value + for key, value in element.attrib.items() + }, + "text": (element.text or "").strip(), + "children": children, + } + + def _strip_namespace(self, value: str) -> str: + if value.startswith("{") and "}" in value: + return value.split("}", 1)[1] + return value + + def _extract_record_path(self, data: Any, record_path: str) -> Any: + current = data + for part in record_path.split("."): + if isinstance(current, dict): + if part not in current: + raise ValidationError( + f"Record path '{record_path}' not found at '{part}'" + ) + current = current[part] + elif isinstance(current, list): + if part.isdigit(): + index = int(part) + try: + current = current[index] + except IndexError as exc: + raise ValidationError( + f"Record path '{record_path}' index out of range: {part}" + ) from exc + else: + values = [] + for item in current: + if not isinstance(item, dict) or part not in item: + raise ValidationError( + f"Record path '{record_path}' not found at '{part}'" + ) + value = item[part] + if isinstance(value, list): + values.extend(value) + else: + values.append(value) + current = values + else: + raise ValidationError( + f"Record path '{record_path}' cannot traverse " + f"{type(current).__name__}" + ) + return current + + def _to_records( + self, data: Any, record_path: Optional[str] = None + ) -> List[Dict[str, Any]]: + selected = self._extract_record_path(data, record_path) if record_path else data + + if isinstance(selected, dict): + for key in ("items", "data", "results", "records"): + value = selected.get(key) + if isinstance(value, list): + selected = value + break + else: + selected = [selected] + elif not isinstance(selected, list): + selected = [selected] + + records: List[Dict[str, Any]] = [] + for item in selected: + if isinstance(item, dict): + records.append(item) + else: + records.append({"value": item}) + return records diff --git a/semantica/ingest/registry.py b/semantica/ingest/registry.py index f24dec90..9d1a9c0a 100644 --- a/semantica/ingest/registry.py +++ b/semantica/ingest/registry.py @@ -13,6 +13,7 @@ Supported Registration Types: * "repo": Repository ingestion methods * "email": Email ingestion methods * "db": Database ingestion methods + * "public_api": Public no-auth API ingestion methods * "parquet": Parquet file and dataset ingestion methods * "ingest": General ingestion methods @@ -58,6 +59,8 @@ class MethodRegistry: "repo": {}, "email": {}, "db": {}, + "api": {}, + "public_api": {}, "mcp": {}, "parquet": {}, "xml": {}, @@ -71,7 +74,8 @@ class MethodRegistry: Args: task: Task type such as "file", "web", "feed", "stream", - "repo", "email", "db", "mcp", "parquet", "xml", or "ingest" + "repo", "email", "db", "public_api", "mcp", "parquet", + "xml", or "ingest" name: Method name method_func: Method function """ @@ -86,7 +90,8 @@ class MethodRegistry: Args: task: Task type such as "file", "web", "feed", "stream", - "repo", "email", "db", "mcp", "parquet", "xml", or "ingest" + "repo", "email", "db", "public_api", "mcp", "parquet", + "xml", or "ingest" name: Method name Returns: @@ -116,7 +121,8 @@ class MethodRegistry: Args: task: Task type such as "file", "web", "feed", "stream", - "repo", "email", "db", "mcp", "parquet", or "ingest" + "repo", "email", "db", "public_api", "mcp", "parquet", + "xml", or "ingest" name: Method name """ if task in cls._methods and name in cls._methods[task]: diff --git a/tests/ingest/test_optional_imports.py b/tests/ingest/test_optional_imports.py index a4e165a1..d040789f 100644 --- a/tests/ingest/test_optional_imports.py +++ b/tests/ingest/test_optional_imports.py @@ -56,6 +56,19 @@ print(FileIngestor.__name__, callable(ingest_file)) assert "FileIngestor True" in result.stdout +def test_public_api_ingestion_imports_without_web_scraping_backends() -> None: + result = _run_python_with_blocked_modules( + """ +from semantica.ingest import PublicAPIIngestor, RESTIngestor, ingest_public_api +print(PublicAPIIngestor.__name__, RESTIngestor.__name__, callable(ingest_public_api)) +""", + ("bs4",), + ) + + assert result.returncode == 0, result.stderr + assert "PublicAPIIngestor RESTIngestor True" in result.stdout + + def test_repository_ingestion_reports_missing_gitpython_when_used() -> None: result = _run_python_with_blocked_modules( """ diff --git a/tests/ingest/test_public_api_ingestor.py b/tests/ingest/test_public_api_ingestor.py new file mode 100644 index 00000000..d76f1549 --- /dev/null +++ b/tests/ingest/test_public_api_ingestor.py @@ -0,0 +1,228 @@ +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from semantica.ingest import ( + APIData, + PublicAPIDetection, + PublicAPIExamples, + PublicAPIIngestor, + ingest, + ingest_public_api, + list_available_methods, +) +from semantica.utils.exceptions import ValidationError + + +def _mock_response( + status_code=200, + json_payload=None, + text="", + headers=None, +): + response = MagicMock() + response.status_code = status_code + response.headers = headers or {} + response.text = text + if json_payload is not None: + response.json.return_value = json_payload + else: + response.json.side_effect = ValueError("not json") + + if status_code >= 400: + response.raise_for_status.side_effect = requests.exceptions.HTTPError( + f"{status_code} error" + ) + else: + response.raise_for_status.return_value = None + return response + + +def test_public_api_examples_catalog_lists_endpoints_and_samples() -> None: + names = PublicAPIExamples.names() + endpoints = PublicAPIExamples.endpoints() + testing_examples = PublicAPIExamples.list_examples(tag="testing") + sample = PublicAPIExamples.sample_response("jsonplaceholder_posts") + + assert "jsonplaceholder_posts" in names + assert endpoints["jsonplaceholder_posts"].startswith("https://") + assert {example.name for example in testing_examples} >= { + "jsonplaceholder_posts", + "jsonplaceholder_users", + "jsonplaceholder_todos", + } + assert sample[0]["title"] == "sample post" + + +def test_public_api_ingestor_ingests_json_records() -> None: + payload = [{"id": 1, "title": "hello"}] + + with patch("requests.Session") as mock_session_class: + mock_session = mock_session_class.return_value + mock_session.headers = {} + mock_session.request.return_value = _mock_response( + json_payload=payload, + text='[{"id": 1, "title": "hello"}]', + headers={"Content-Type": "application/json"}, + ) + + result = PublicAPIIngestor(rate_limit_delay=0).ingest_public_api( + "https://jsonplaceholder.typicode.com/posts" + ) + + assert isinstance(result, APIData) + assert result.data == payload + assert result.metadata["public_api"] is True + assert result.metadata["authentication"] == "none" + assert result.metadata["response_format"] == "json" + assert result.metadata["record_count"] == 1 + + +def test_public_api_ingestor_extracts_nested_record_path() -> None: + payload = { + "success": True, + "result": { + "count": 1, + "results": [{"id": "dataset-1", "title": "Dataset"}], + }, + } + + with patch("requests.Session") as mock_session_class: + mock_session = mock_session_class.return_value + mock_session.headers = {} + mock_session.request.return_value = _mock_response( + json_payload=payload, + text='{"success": true}', + headers={"Content-Type": "application/json"}, + ) + + result = PublicAPIIngestor(rate_limit_delay=0).ingest_public_api( + "https://catalog.data.gov/api/3/action/package_search", + record_path="result.results", + ) + + assert result.data == [{"id": "dataset-1", "title": "Dataset"}] + assert result.metadata["record_path"] == "result.results" + + +def test_public_api_ingestor_parses_csv_records() -> None: + csv_text = "id,name\n1,Ada\n2,Grace\n" + + with patch("requests.Session") as mock_session_class: + mock_session = mock_session_class.return_value + mock_session.headers = {} + mock_session.request.return_value = _mock_response( + text=csv_text, + headers={"Content-Type": "text/csv"}, + ) + + result = PublicAPIIngestor(rate_limit_delay=0).ingest_public_api( + "https://example.com/data.csv" + ) + + assert result.data == [{"id": "1", "name": "Ada"}, {"id": "2", "name": "Grace"}] + assert result.metadata["response_format"] == "csv" + + +def test_public_api_ingestor_parses_xml_records_with_record_path() -> None: + xml_text = "AdaGrace" + + with patch("requests.Session") as mock_session_class: + mock_session = mock_session_class.return_value + mock_session.headers = {} + mock_session.request.return_value = _mock_response( + text=xml_text, + headers={"Content-Type": "application/xml"}, + ) + + result = PublicAPIIngestor(rate_limit_delay=0).ingest_public_api( + "https://example.com/data.xml", + record_path="children", + ) + + assert result.metadata["response_format"] == "xml" + assert result.data[0]["tag"] == "item" + assert result.data[0]["attributes"] == {"id": "1"} + assert result.data[0]["text"] == "Ada" + + +def test_public_api_detection_reports_public_endpoint() -> None: + with patch("requests.Session") as mock_session_class: + mock_session = mock_session_class.return_value + mock_session.headers = {} + mock_session.request.return_value = _mock_response( + headers={"Content-Type": "application/json"} + ) + + detection = PublicAPIIngestor(rate_limit_delay=0).detect_public_api( + "https://jsonplaceholder.typicode.com/posts" + ) + + assert isinstance(detection, PublicAPIDetection) + assert detection.is_public is True + assert detection.requires_auth is False + assert detection.response_status == 200 + + +def test_public_api_detection_reports_auth_required() -> None: + with patch("requests.Session") as mock_session_class: + mock_session = mock_session_class.return_value + mock_session.headers = {} + mock_session.request.return_value = _mock_response( + status_code=401, + headers={"WWW-Authenticate": "Bearer"}, + ) + + detection = PublicAPIIngestor(rate_limit_delay=0).detect_public_api( + "https://api.example.com/private" + ) + + assert detection.is_public is False + assert detection.requires_auth is True + assert detection.response_status == 401 + + +def test_public_api_ingestor_rejects_authentication_inputs() -> None: + with patch("requests.Session") as mock_session_class: + mock_session = mock_session_class.return_value + mock_session.headers = {} + ingestor = PublicAPIIngestor(rate_limit_delay=0) + + with pytest.raises(ValidationError, match="no-auth endpoints"): + ingestor.ingest_public_api( + "https://api.example.com/data", + headers={"Authorization": "Bearer token"}, + ) + + mock_session.request.assert_not_called() + + +def test_public_api_convenience_methods_and_registry_dispatch() -> None: + payload = [{"id": 1}] + + with patch("requests.Session") as mock_session_class: + mock_session = mock_session_class.return_value + mock_session.headers = {} + mock_session.request.return_value = _mock_response( + json_payload=payload, + text='[{"id": 1}]', + headers={"Content-Type": "application/json"}, + ) + + direct = ingest_public_api( + "https://jsonplaceholder.typicode.com/posts", + rate_limit_delay=0, + ) + unified = ingest( + "https://jsonplaceholder.typicode.com/posts", + source_type="public_api", + rate_limit_delay=0, + ) + methods = list_available_methods("public_api") + + assert isinstance(direct, APIData) + assert direct.data == payload + assert unified["data"].data == payload + assert "endpoint" in methods["public_api"] + assert "detect" in methods["public_api"]