-
Notifications
You must be signed in to change notification settings - Fork 0
/
Event.py
150 lines (125 loc) · 4.77 KB
/
Event.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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
import asyncio
import datetime
import sys
from typing import Literal
import aiohttp
import logging
from HelperFunctions import fetch_json
from config import API_ENDPOINT
class Event:
def __init__(
self,
id: int,
title: str = None,
start: str = None,
end: str = None,
deadline: str = None,
signup_start: str = None,
place: str = None,
status: Literal["EXPIRED", "CLOSED", "ACTIVE", "NO_SIGNUP", "TBA"] = None,
):
"""
Creates a new Event given a set of parameters
"""
# Sets fields to the input values and converts iso-strings to datetime
self.id = id
self.title = title
self.start = None if start is None else datetime.datetime.fromisoformat(start)
self.end = None if end is None else datetime.datetime.fromisoformat(end)
self.deadline = (
None if deadline is None else datetime.datetime.fromisoformat(deadline)
)
self.signup_start = (
None
if signup_start is None
else datetime.datetime.fromisoformat(signup_start)
)
self.place = place
self.status = status
def to_json(self) -> dict:
"""Returns a dict representation of an event"""
logging.debug(f"Convertet event with id: {self.id}, to dict")
# Copies dict so to not make changes to the object (mutable)
result_json = self.__dict__.copy()
# Converts all date fields to a string representation
for key, value in result_json.items():
if type(value) == datetime.datetime:
result_json[key] = value.isoformat()
return result_json
@classmethod
async def get_event(cls, session: aiohttp.ClientSession, id: int) -> "Event":
url = f"{API_ENDPOINT}{id}"
event_json = await fetch_json(session, url)
# Bad response
# Event(id) creates an event where id is the value of id and all other fields are set to None
if len(event_json) == 1:
logging.warning(
f"The request to evnt with url {url} had a response with length 1"
)
return Event(id)
try:
title = event_json["title"]
start_datetime = event_json["start_date"]
end_datetime = event_json["end_date"]
deadline = event_json["end_registration_at"]
signup_start = event_json["start_registration_at"]
place = event_json["location"]
if event_json["expired"]:
status = "EXPIRED"
elif event_json["closed"]:
status = "CLOSED"
elif event_json["sign_up"]:
status = "ACTIVE"
elif event_json["description"] == "TBA":
status = "TBA"
else:
status = "NO_SIGNUP"
# Bad JSON
except KeyError as e:
logging.warning(
f"Something was from with the json returned from the request [url: {url}]. KeyError: '{e}'"
)
return Event(id)
# Creating new event with the fetched parameters
event = Event(
id=id,
title=title,
start=start_datetime,
end=end_datetime,
deadline=deadline,
signup_start=signup_start,
place=place,
status=status,
)
logging.debug(f"Event with id {event.id} retrived")
return event
def copy(self) -> "Event":
"""Returns a copy of the instance of `Event`"""
logging.debug(f"Copied event with id: {self.id}")
return self.__class__(**self.to_json())
def __eq__(self, other: "Event"):
return all(
self.__dict__[key] == other.__dict__[key] for key in self.__dict__.keys()
)
def __repr__(self):
logging.debug(f"Returned string representation of event with id: {self.id}")
# Gets the json repr of the class (all strings instead of datetime fields)
dic = self.to_json()
# Formatted as you would call the function
return f"{type(self).__name__}({', '.join(f'{key}={value}' for key, value in dic.items())})"
if __name__ == "__main__":
# To remove RuntimeError on exit on windows as documented in this issue:
# https://github.com/encode/httpx/issues/914
if (
sys.version_info[0] == 3
and sys.version_info[1] >= 8
and sys.platform.startswith("win")
):
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
# Troubleshooting:
async def main():
# async with aiohttp.ClientSession() as session:
# print(await Event.get_event(session, 277))
return
# logging.basicConfig(filename="test.log", level=logging.DEBUG)
asyncio.run(main())