File: error_handling.py

package info (click to toggle)
python-osmapi 4.3.0%2Bds-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 820 kB
  • sloc: python: 3,409; xml: 1,601; makefile: 45; sh: 14
file content (144 lines) | stat: -rw-r--r-- 4,622 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
from dotenv import load_dotenv, find_dotenv
from oauthcli import OpenStreetMapDevAuth
from oauthlib.oauth2.rfc6749.errors import OAuth2Error
import logging
import os
import osmapi
import requests
import subprocess
import sys
import urllib3


load_dotenv(find_dotenv())

# logging setup
log = logging.getLogger(__name__)
loglevel = logging.DEBUG
logging.basicConfig(
    format="%(asctime)s %(name)s %(levelname)-8s %(message)s",
    level=loglevel,
    datefmt="%Y-%m-%d %H:%M:%S",
)
logging.captureWarnings(True)

# shut up DEBUG messages of specific loggers
logging.getLogger(osmapi.dom.__name__).setLevel(logging.INFO)
logging.getLogger(urllib3.__name__).setLevel(logging.INFO)


def clear_screen():
    # check and make call for specific operating system
    _ = subprocess.call("clear" if os.name == "posix" else "cls")


# The error handling with osmapi is very easy, simply catch the
# exception for the specific case you want to handle.
# - All osmapi excepctions are child classes of osmapi.OsmApiError
# - Errors that result from the communication with the OSM server osmapi.ApiError
# - There are a number of subclasses to differantiate the different errors
# - catch more specific errors first, then use more general error classes

# Upload data to OSM without a changeset
log.debug("Try to write data to OSM without a changeset")
api = osmapi.OsmApi(api="https://api06.dev.openstreetmap.org")
try:
    node1 = api.NodeCreate({"lon": 1, "lat": 1, "tag": {}})
except osmapi.NoChangesetOpenError as e:
    log.exception(e)
    log.debug("There is no open changeset")
input("Press Enter to continue...")
clear_screen()


# wrong server: ConnectionError
log.debug("Connect to wrong server...")
api = osmapi.OsmApi(api="https://invalid.server.name")
try:
    api.ChangesetGet(123)
except osmapi.ConnectionApiError as e:
    log.exception(e)
    log.debug("Error connecting to server")
input("Press Enter to continue...")
clear_screen()


# changeset not found: ElementNotFoundApiError
log.debug("Request non-existent changeset id...")
api = osmapi.OsmApi(api="https://api06.dev.openstreetmap.org")
try:
    api.ChangesetGet(111111111111)
except osmapi.ElementNotFoundApiError as e:
    log.exception(e)
    log.debug("Changeset not found")
input("Press Enter to continue...")
clear_screen()


# unauthorized request
log.debug("Try to add data with wrong authorization")
try:
    s = requests.Session()
    s.auth = ("user", "pass")
    api = osmapi.OsmApi(api="https://api06.dev.openstreetmap.org", session=s)
    with api.Changeset({"comment": "My first test"}) as changeset_id:
        node1 = api.NodeCreate({"lon": 1, "lat": 1, "tag": {}})
except osmapi.UnauthorizedApiError as e:
    log.exception(e)
    log.debug("Unauthorized to make this request")
input("Press Enter to continue...")
clear_screen()

# request without auhorization
log.debug("Try to add data without authorization")
try:
    api = osmapi.OsmApi(api="https://api06.dev.openstreetmap.org")
    with api.Changeset({"comment": "My first test"}) as changeset_id:
        node1 = api.NodeCreate({"lon": 1, "lat": 1, "tag": {}})
except osmapi.UsernamePasswordMissingError as e:
    log.exception(e)
    log.debug("Username/Password or authorization missing")
input("Press Enter to continue...")
clear_screen()


# a more or less complete "real-life" example
client_id = os.getenv("OSM_OAUTH_CLIENT_ID")
client_secret = os.getenv("OSM_OAUTH_CLIENT_SECRET")

try:
    auth = OpenStreetMapDevAuth(
        client_id, client_secret, ["write_api", "write_notes"]
    ).auth_code()
except OAuth2Error as e:
    log.exception(e)
    log.debug("An OAuth2 error occured")
    sys.exit(1)

try:
    api = osmapi.OsmApi(api="https://api06.dev.openstreetmap.org", session=auth.session)
    with api.Changeset({"comment": "My first test"}) as changeset_id:
        log.debug(f"Part of Changeset {changeset_id}")
        node1 = api.NodeCreate({"lon": 1, "lat": 1, "tag": {}})
        log.debug(node1)

    # get all the info from the closed changeset
    changeset = api.ChangesetGet(changeset_id)
    log.debug(changeset)
    exit_code = 0
except osmapi.ConnectionApiError as e:
    log.debug(f"Connection error: {str(e)}")
    exit_code = 1
    # display error for user, try again?
except osmapi.ElementNotFoundApiError as e:
    log.debug(f"Changeset not found: {str(e)}")
    exit_code = 1
except osmapi.ApiError as e:
    log.debug(f"Error on the API side: {str(e)}")
    exit_code = 1
except osmapi.OsmApiError as e:
    log.debug(f"Some other error: {str(e)}")
    exit_code = 1
finally:
    log.debug(f"Exit code: {exit_code}")
    sys.exit(exit_code)