Error Handling

class OsmApiError(Exception):
    """
    General OsmApi error class serving as superclass for all other errors.
    Base exception for all osmapi-specific errors.
    """

class ApiError(OsmApiError):
    """
    Base API request error with detailed response information.
    
    Attributes:
    - status (int): HTTP error code
    - reason (str): Error message
    - payload (str): Response payload when error occurred
    """
    
    def __init__(self, status, reason, payload):
        self.status = status
        self.reason = reason  
        self.payload = payload

import osmapi

api = osmapi.OsmApi()

try:
    node = api.NodeGet(999999)
except osmapi.ApiError as e:
    print(f"API Error {e.status}: {e.reason}")
    if e.payload:
        print(f"Server response: {e.payload}")
except osmapi.OsmApiError as e:
    print(f"General osmapi error: {e}")

class UsernamePasswordMissingError(OsmApiError):
    """
    Error when username or password is missing for an authenticated request.
    
    Raised when attempting operations that require authentication
    without providing credentials.
    """

import osmapi

# No authentication provided
api = osmapi.OsmApi()

try:
    # This requires authentication
    api.ChangesetCreate({"comment": "Test changeset"})
except osmapi.UsernamePasswordMissingError:
    print("Authentication required for this operation")
    # Provide credentials
    api = osmapi.OsmApi(username="user", password="pass")

class UnauthorizedApiError(ApiError):
    """
    Error when the API returned an Unauthorized error.
    
    Raised when provided OAuth token is expired, invalid credentials
    are used, or account lacks permissions for the operation.
    """

import osmapi

api = osmapi.OsmApi(username="wrong_user", password="wrong_pass")

try:
    api.ChangesetCreate({"comment": "Test"})
except osmapi.UnauthorizedApiError as e:
    print(f"Authentication failed: {e.reason}")
    # Check credentials or OAuth token

class TimeoutApiError(ApiError):
    """
    Error if the HTTP request ran into a timeout.
    
    Raised when requests exceed the configured timeout period.
    """

class ConnectionApiError(ApiError):
    """
    Error if there was a network error while connecting to the remote server.
    
    Includes DNS failures, refused connections, and other network issues.
    """

class MaximumRetryLimitReachedError(OsmApiError):
    """
    Error when the maximum amount of retries is reached and we have to give up.
    
    Raised when all retry attempts are exhausted for server errors.
    """

import osmapi

api = osmapi.OsmApi(timeout=5)  # Short timeout for example

try:
    node = api.NodeGet(123)
except osmapi.TimeoutApiError as e:
    print(f"Request timed out: {e.reason}")
    # Retry with longer timeout or check network
except osmapi.ConnectionApiError as e:
    print(f"Network error: {e.reason}")
    # Check internet connection
except osmapi.MaximumRetryLimitReachedError as e:
    print(f"Server unavailable after retries: {e}")

class XmlResponseInvalidError(OsmApiError):
    """
    Error if the XML response from the OpenStreetMap API is invalid.
    
    Raised when the server returns malformed XML or unexpected response format.
    """

class ResponseEmptyApiError(ApiError):
    """
    Error when the response to the request is empty.
    
    Raised when expecting data but receiving empty response from server.
    """

import osmapi

api = osmapi.OsmApi()

try:
    nodes = api.NodesGet([123, 456])
except osmapi.XmlResponseInvalidError as e:
    print(f"Invalid XML response: {e}")
    # Server may be experiencing issues
except osmapi.ResponseEmptyApiError as e:
    print(f"Empty response: {e.reason}")
    # May indicate server problems or invalid request

class NoChangesetOpenError(OsmApiError):
    """
    Error when an operation requires an open changeset, but currently
    no changeset is open.
    
    Raised when attempting data modifications without an active changeset.
    """

class ChangesetAlreadyOpenError(OsmApiError):
    """
    Error when a user tries to open a changeset when there is already
    an open changeset.
    
    Only one changeset can be open at a time per user session.
    """

class ChangesetClosedApiError(ApiError):
    """
    Error if the changeset in question has already been closed.
    
    Raised when attempting to modify a changeset that is no longer active.
    """

import osmapi

api = osmapi.OsmApi(username="user", password="pass")

try:
    # Forgot to open changeset
    api.NodeCreate({"lat": 0, "lon": 0, "tag": {}})
except osmapi.NoChangesetOpenError:
    print("Need to open a changeset first")
    changeset_id = api.ChangesetCreate({"comment": "Adding nodes"})

try:
    # Try to open second changeset
    api.ChangesetCreate({"comment": "Another changeset"})
except osmapi.ChangesetAlreadyOpenError:
    print("Close current changeset before opening new one")
    api.ChangesetClose()

class ElementNotFoundApiError(ApiError):
    """
    Error if the requested element was not found.
    
    Raised when requesting an element ID that doesn't exist.
    """

class ElementDeletedApiError(ApiError):
    """
    Error when the requested element is deleted.
    
    Raised when requesting an element that has been marked as deleted.
    """

class OsmTypeAlreadyExistsError(OsmApiError):
    """
    Error when a user tries to create an object that already exists.
    
    Raised when attempting to create an element with an existing ID.
    """

import osmapi

api = osmapi.OsmApi()

try:
    node = api.NodeGet(999999999)
except osmapi.ElementNotFoundApiError:
    print("Node doesn't exist")
except osmapi.ElementDeletedApiError:
    print("Node has been deleted")

# Creating elements
api_auth = osmapi.OsmApi(username="user", password="pass")
try:
    with api_auth.Changeset({"comment": "Test"}) as changeset_id:
        # This would fail if trying to create with existing ID
        existing_node = {"id": 123, "lat": 0, "lon": 0, "tag": {}}
        api_auth.NodeCreate(existing_node)
except osmapi.OsmTypeAlreadyExistsError:
    print("Cannot create element with existing ID")

class VersionMismatchApiError(ApiError):
    """
    Error if the provided version does not match the database version
    of the element.
    
    Raised when attempting to modify an element that has been changed
    by another user since it was last retrieved.
    """

class PreconditionFailedApiError(ApiError):
    """
    Error if the precondition of the operation was not met.
    
    Raised when:
    - A way has nodes that do not exist or are not visible
    - A relation has elements that do not exist or are not visible  
    - A node/way/relation is still used in a way/relation when deleting
    """

import osmapi

api = osmapi.OsmApi(username="user", password="pass")

try:
    # Get current node
    node = api.NodeGet(12345)
    
    with api.Changeset({"comment": "Update node"}) as changeset_id:
        # Someone else might have modified it meanwhile
        node["tag"]["updated"] = "yes"
        api.NodeUpdate(node)
        
except osmapi.VersionMismatchApiError as e:
    print(f"Version conflict: {e.reason}")
    # Re-fetch current version and merge changes
    current_node = api.NodeGet(12345)
    # Apply changes to current version

try:
    with api.Changeset({"comment": "Create way"}) as changeset_id:
        # This will fail if nodes don't exist
        api.WayCreate({
            "nd": [999999, 999998],  # Non-existent nodes
            "tag": {"highway": "path"}
        })
except osmapi.PreconditionFailedApiError as e:
    print(f"Referenced elements don't exist: {e.reason}")

class NoteAlreadyClosedApiError(ApiError):
    """
    Error if the note in question has already been closed.
    
    Raised when attempting to close a note that is already closed.
    """

import osmapi

api = osmapi.OsmApi(username="user", password="pass")

try:
    api.NoteClose(12345, "Issue resolved")
except osmapi.NoteAlreadyClosedApiError:
    print("Note is already closed")
    # Maybe reopen it instead
    api.NoteReopen(12345, "Issue has returned")

class AlreadySubscribedApiError(ApiError):
    """
    Error when a user tries to subscribe to a changeset
    that they are already subscribed to.
    """

class NotSubscribedApiError(ApiError):
    """
    Error when user tries to unsubscribe from a changeset
    that they are not subscribed to.
    """

import osmapi

api = osmapi.OsmApi(username="user", password="pass")

try:
    api.ChangesetSubscribe(12345)
except osmapi.AlreadySubscribedApiError:
    print("Already subscribed to this changeset")

try:
    api.ChangesetUnsubscribe(12345)
except osmapi.NotSubscribedApiError:
    print("Not subscribed to this changeset")

import osmapi
import time

def robust_node_get(api, node_id, max_retries=3):
    """Get node with retry logic and graceful error handling."""
    
    for attempt in range(max_retries):
        try:
            return api.NodeGet(node_id)
            
        except osmapi.ElementNotFoundApiError:
            # Element doesn't exist, no point retrying
            return None
            
        except osmapi.ElementDeletedApiError:
            # Element was deleted, return None or handle specially
            return {"deleted": True, "id": node_id}
            
        except (osmapi.TimeoutApiError, osmapi.ConnectionApiError):
            if attempt < max_retries - 1:
                # Wait before retry
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            else:
                # Final attempt failed
                raise
                
        except osmapi.MaximumRetryLimitReachedError:
            # Server is having issues
            print(f"Server unavailable, giving up on node {node_id}")
            return None
            
    return None

import osmapi

def update_with_conflict_resolution(api, element_type, element_data):
    """Update element with automatic conflict resolution."""
    
    max_attempts = 3
    
    for attempt in range(max_attempts):
        try:
            if element_type == "node":
                return api.NodeUpdate(element_data)
            elif element_type == "way":
                return api.WayUpdate(element_data)
            elif element_type == "relation":
                return api.RelationUpdate(element_data)
                
        except osmapi.VersionMismatchApiError:
            if attempt < max_attempts - 1:
                # Fetch current version
                element_id = element_data["id"]
                
                if element_type == "node":
                    current = api.NodeGet(element_id)
                elif element_type == "way":
                    current = api.WayGet(element_id)  
                elif element_type == "relation":
                    current = api.RelationGet(element_id)
                
                # Merge changes (simple example - merge tags)
                current["tag"].update(element_data["tag"])
                element_data = current
                continue
            else:
                # Too many conflicts, give up
                raise
                
    return None

import osmapi
import logging

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def safe_osm_operation(operation_func, *args, **kwargs):
    """Wrapper for OSM operations with comprehensive error logging."""
    
    try:
        result = operation_func(*args, **kwargs)
        logger.info(f"Operation successful: {operation_func.__name__}")
        return result
        
    except osmapi.UsernamePasswordMissingError as e:
        logger.error(f"Authentication required: {e}")
        raise
        
    except osmapi.UnauthorizedApiError as e:
        logger.error(f"Authentication failed: {e.status} - {e.reason}")
        raise
        
    except osmapi.ElementNotFoundApiError as e:
        logger.warning(f"Element not found: {e.status} - {e.reason}")
        return None
        
    except osmapi.VersionMismatchApiError as e:
        logger.warning(f"Version conflict: {e.status} - {e.reason}")
        # Could implement retry logic here
        raise
        
    except osmapi.ChangesetClosedApiError as e:
        logger.error(f"Changeset closed: {e.status} - {e.reason}")
        raise
        
    except osmapi.TimeoutApiError as e:
        logger.error(f"Request timeout: {e.reason}")
        # Could implement retry logic
        raise
        
    except osmapi.ApiError as e:
        logger.error(f"API Error {e.status}: {e.reason}")
        if e.payload:
            logger.error(f"Server response: {e.payload}")
        raise
        
    except osmapi.OsmApiError as e:
        logger.error(f"OSM API Error: {e}")
        raise
        
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise

# Usage example
api = osmapi.OsmApi(username="user", password="pass")

# Safe operation with logging
result = safe_osm_operation(api.NodeGet, 12345)
if result:
    print(f"Got node: {result['id']}")

def validate_node_data(node_data):
    """Validate node data before API calls."""
    
    required_fields = ["lat", "lon"]
    for field in required_fields:
        if field not in node_data:
            raise ValueError(f"Missing required field: {field}")
    
    # Validate coordinate ranges
    if not -90 <= node_data["lat"] <= 90:
        raise ValueError(f"Invalid latitude: {node_data['lat']}")
    
    if not -180 <= node_data["lon"] <= 180:
        raise ValueError(f"Invalid longitude: {node_data['lon']}")
    
    # Validate tags
    if "tag" in node_data:
        for key, value in node_data["tag"].items():
            if not isinstance(key, str) or not isinstance(value, str):
                raise ValueError("Tag keys and values must be strings")

import osmapi

class SafeChangesetManager:
    """Context manager with enhanced error handling."""
    
    def __init__(self, api, changeset_tags):
        self.api = api
        self.changeset_tags = changeset_tags
        self.changeset_id = None
    
    def __enter__(self):
        try:
            self.changeset_id = self.api.ChangesetCreate(self.changeset_tags)
            return self.changeset_id
        except osmapi.ChangesetAlreadyOpenError:
            # Close existing changeset and try again
            self.api.ChangesetClose()
            self.changeset_id = self.api.ChangesetCreate(self.changeset_tags)
            return self.changeset_id
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.changeset_id:
            try:
                self.api.ChangesetClose()
            except osmapi.ChangesetClosedApiError:
                # Already closed, that's fine
                pass
            except Exception as e:
                logger.error(f"Error closing changeset: {e}")

# Usage
api = osmapi.OsmApi(username="user", password="pass")
with SafeChangesetManager(api, {"comment": "Safe operations"}) as changeset_id:
    # Perform operations safely
    pass