如何在SPARQLWrapper中获取错误码并捕获端点崩溃异常？

I've run into exactly this frustrating issue before—SPARQLWrapper's default behavior doesn't throw an exception when the endpoint crashes on the first query; it just returns an empty result set, and you only get a connection refused error on the next attempt. That makes it impossible to pinpoint the actual crash at the moment it happens. Here are a few approaches I've used to fix this:

1. Manually Check the HTTP Response Status Code #

Instead of calling convert() right away, first fetch the raw response object and validate its status code. This lets you catch server errors (like 5xx codes) immediately:

from SPARQLWrapper import SPARQLWrapper, JSON
from urllib.error import HTTPError, URLError

sparql = SPARQLWrapper("your-endpoint-url")
sparql.setQuery("your-sparql-query")
sparql.setReturnFormat(JSON)

try:
    response = sparql.query()
    # Verify the HTTP status is successful
    if response.status != 200:
        raise Exception(f"Endpoint error: HTTP status {response.status}")
    
    results = response.convert()
    # Add an extra check if your query shouldn't return empty results
    if not results.get("results", {}).get("bindings"):
        # Adjust this based on your query type (e.g., SELECT vs ASK)
        raise Exception("Empty result set detected—endpoint may have crashed")
except HTTPError as e:
    print(f"HTTP error occurred: {e.code} - {e.reason}")
except URLError as e:
    print(f"Connection failed: {e.reason}")
except Exception as e:
    print(f"Query error: {str(e)}")

The QueryResult object returned by sparql.query() exposes a status property with the HTTP response code. If the endpoint crashes mid-request, it'll often return a 5xx status that you can catch here.

2. Catch Low-Level Request Exceptions #

SPARQLWrapper uses urllib under the hood (or requests in newer versions), so you can catch the specific exceptions from those libraries to handle connection issues and server errors:

from SPARQLWrapper import SPARQLWrapper, JSON
from urllib.error import HTTPError, URLError

sparql = SPARQLWrapper("your-endpoint-url")
sparql.setQuery("your-sparql-query")
sparql.setReturnFormat(JSON)
# Optional: Set a timeout to avoid hanging on unresponsive endpoints
sparql.setTimeout(10)

try:
    results = sparql.query().convert()
    # Validate non-empty results for SELECT queries
    if query.strip().upper().startswith("SELECT") and not results.get("results", {}).get("bindings"):
        raise RuntimeError("SELECT query returned no data—endpoint might have crashed")
except HTTPError as e:
    print(f"Server returned error: {e.code} - {e.reason}")
except URLError as e:
    print(f"Could not connect to endpoint: {e.reason}")
except RuntimeError as e:
    print(f"Query validation failed: {str(e)}")

Adding a timeout ensures that if the endpoint becomes unresponsive, you won't wait indefinitely—you'll get an exception right away.

3. Wrap Queries in a Custom Utility Function #

For reusability, wrap all your SPARQL logic in a helper function that handles status checks, empty result validation, and exception handling consistently:

from SPARQLWrapper import SPARQLWrapper, JSON
from urllib.error import HTTPError, URLError

def execute_sparql(endpoint_url, query, timeout=10):
    sparql = SPARQLWrapper(endpoint_url)
    sparql.setQuery(query)
    sparql.setReturnFormat(JSON)
    sparql.setTimeout(timeout)
    
    try:
        response = sparql.query()
        if response.status != 200:
            raise RuntimeError(f"Endpoint returned status code {response.status}")
        
        results = response.convert()
        # Handle different query types
        query_type = query.strip().upper().split()[0]
        if query_type == "SELECT" and not results.get("results", {}).get("bindings"):
            raise RuntimeError("SELECT query returned empty bindings—potential endpoint crash")
        elif query_type == "ASK" and not results.get("boolean"):
            # ASK queries return a boolean; adjust this check based on your expectations
            pass
        
        return results
    except (HTTPError, URLError) as e:
        raise RuntimeError(f"Connection/HTTP error: {str(e)}") from e
    except Exception as e:
        raise RuntimeError(f"Query failed: {str(e)}") from e

# Usage example
try:
    results = execute_sparql("your-endpoint-url", "SELECT ?s ?p ?o WHERE { ?s ?p ?o } LIMIT 10")
    print(f"Found {len(results['results']['bindings'])} results")
except RuntimeError as e:
    print(f"Error executing query: {e}")

This function adapts its validation based on the query type, so you don't false-positive on ASK queries that return false or CONSTRUCT queries that return empty graphs when expected.

内容的提问来源于stack exchange，提问作者Yu Gu