-
Notifications
You must be signed in to change notification settings - Fork 0
/
metmuseum.py
124 lines (86 loc) · 2.85 KB
/
metmuseum.py
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
from sys import exit
import datetime
try:
import httpx
except (ModuleNotFoundError, ImportError):
print("Les dependances n'ont pas ete installees, 'pip install -r requirements.txt'")
exit(1)
KEYWORDS = ["q", "isHighlight", "title", "tags", "departmentId", "isOnView", "artistOrCulture",
"medium", "hasImages", "geoLocation", "dateBegin", "dateEnd"]
BASE_URL = "https://collectionapi.metmuseum.org/public/collection/v1/"
CLIENT = httpx.AsyncClient()
async def close_session():
await CLIENT.aclose()
async def fetch_objects(metadataDate=None, departmentIds=None):
url = BASE_URL+"objects?"
if not metadataDate is None:
try:
datetime.date.fromisoformat(metadataDate)
except ValueError as e:
raise ValueError(str(e)+", date format must be YYYY-MM-DD.")
url += "metadataDate="+metadataDate
if not departmentIds is None:
url += "&"
elif not departmentIds is None:
url += "departmentIds"+"="
if type(departmentIds) in (tuple, list):
for i in departmentIds:
url += str(i)+"|"
url = url[:-1] # Enlever le dernier "|"
else:
url += str(departmentIds)
if url[-1]=="&":
url = url[:-1]
resp = await CLIENT.get(url)
return resp.json()
# Je ne renvoie pas de message d'erreur si l'id n'est pas bon car
# la requete search renvoie de faux ids. Je ne peux pas utiliser
# une exception pour gerer chaque requete car elles sont envoyees
# en meme temps.
async def fetch_object(id):
resp = await CLIENT.get(BASE_URL+f"objects/{str(id)}")
return resp.json()
async def fetch_departments(department=None):
resp = await CLIENT.get(BASE_URL+"departments")
resp = resp.json()
res = {"departments":[]}
try:
dep = resp["departments"]
for i in dep:
id = i["departmentId"]
if type(department) in (list, tuple):
if id in department:
res["departments"].append(i)
elif type(department)==int:
if id==department:
res["departments"].append(i)
break
except KeyError:
raise ValueError(f"{str(department)} is/are not valid id/s.")
if len(res["departments"])==0:
raise ValueError(f"{str(department)} is/are not valid id/s.")
return res
async def fetch(**kwargs):
url = BASE_URL+"search?"
if "q" not in kwargs.keys():
raise ValueError("Query must have a q parameter.")
for i in kwargs:
if i not in KEYWORDS:
raise ValueError(f"{i} is not a valid keyword.")
if i=="medium" or i=="geoLocation":
url += i+"="
if type(kwargs[i]) in (list, tuple):
for j in kwargs[i]:
url += str(kwargs[i])+"|"
url = url[:-1]
else:
url += str(kwargs[i])
elif type(kwargs[i])==bool:
kwargs[i] = str(kwargs[i]).lower()
else:
url += i+"="+str(kwargs[i])
url += "&"
if url[-1]=="&":
url = url[:-1]
resp = await CLIENT.get(url)
return resp.json()