Digitale bierlijst

model.py 8.0KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276
  1. """
  2. Provides access to the models stored in the database, via the server.
  3. """
  4. import datetime
  5. from typing import NamedTuple
  6. from urllib.parse import urljoin
  7. import requests
  8. from . import logger
  9. LOG = logger.getLogger("model")
  10. SERVER_URL = "http://127.0.0.1:5000"
  11. class Person(NamedTuple):
  12. """ Represents a Person, as retrieved from the database. """
  13. name: str
  14. person_id: int = None
  15. consumptions: dict = {}
  16. def add_consumption(self, type_id: str) -> bool:
  17. """ Register a consumption for this Person. """
  18. req = requests.post(
  19. urljoin(SERVER_URL, f"people/{self.person_id}/add_consumption/{type_id}")
  20. )
  21. try:
  22. data = req.json()
  23. if "error" in data:
  24. LOG.error(
  25. "Could not add consumption for %s (%s): %s",
  26. self.person_id,
  27. req.status_code,
  28. data,
  29. )
  30. return False
  31. self.consumptions.update(data["person"]["consumptions"])
  32. return Consumption.from_dict(data["consumption"])
  33. except ValueError:
  34. LOG.error(
  35. "Did not get JSON on adding Consumption (%s): %s",
  36. req.status_code,
  37. req.content,
  38. )
  39. return False
  40. def create(self) -> "Person":
  41. """ Create a new Person from the current attributes. As tuples are
  42. immutable, a new Person with the correct id is returned. """
  43. req = requests.post(
  44. urljoin(SERVER_URL, "people"), json={"person": {"name": self.name}}
  45. )
  46. try:
  47. data = req.json()
  48. except ValueError:
  49. LOG.error(
  50. "Did not get JSON on adding Person (%s): %s",
  51. req.status_code,
  52. req.content,
  53. )
  54. return None
  55. if "error" in data or req.status_code != 201:
  56. LOG.error("Could not create Person (%s): %s", req.status_code, data)
  57. return None
  58. return Person.from_dict(data["person"])
  59. @classmethod
  60. def get(cls, person_id: int) -> "Person":
  61. """ Retrieve a Person by id. """
  62. req = requests.get(urljoin(SERVER_URL, f"/people/{person_id}"))
  63. try:
  64. data = req.json()
  65. if "error" in data:
  66. LOG.warning(
  67. "Could not get person %s (%s): %s", person_id, req.status_code, data
  68. )
  69. return None
  70. return Person.from_dict(data["person"])
  71. except ValueError:
  72. LOG.error(
  73. "Did not get JSON from server on getting Person (%s): %s",
  74. req.status_code,
  75. req.content,
  76. )
  77. return None
  78. @classmethod
  79. def get_all(cls, active=None) -> ["Person"]:
  80. """ Get all active People. """
  81. req = requests.get(urljoin(SERVER_URL, "/people"),
  82. params={'active':active})
  83. try:
  84. data = req.json()
  85. if "error" in data:
  86. LOG.warning("Could not get people (%s): %s", req.status_code, data)
  87. return [Person.from_dict(item) for item in data["people"]]
  88. except ValueError:
  89. LOG.error(
  90. "Did not get JSON from server on getting People (%s): %s",
  91. req.status_code,
  92. req.content,
  93. )
  94. return None
  95. @classmethod
  96. def from_dict(cls, data: dict) -> "Person":
  97. """ Reconstruct a Person object from a dict. """
  98. return Person(
  99. name=data["name"],
  100. person_id=data["person_id"],
  101. consumptions=data["consumptions"],
  102. )
  103. class ConsumptionType(NamedTuple):
  104. """ Represents a stored ConsumptionType. """
  105. name: str
  106. consumption_type_id: int = None
  107. icon: str = None
  108. def create(self) -> "ConsumptionType":
  109. """ Create a new ConsumptionType from the current attributes. As tuples
  110. are immutable, a new ConsumptionType with the correct id is returned.
  111. """
  112. req = requests.post(
  113. urljoin(SERVER_URL, "consumption_types"),
  114. json={"consumption_type": {"name": self.name, "icon": self.icon}},
  115. )
  116. try:
  117. data = req.json()
  118. except ValueError:
  119. LOG.error(
  120. "Did not get JSON on adding ConsumptionType (%s): %s",
  121. req.status_code,
  122. req.content,
  123. )
  124. return None
  125. if "error" in data or req.status_code != 201:
  126. LOG.error(
  127. "Could not create ConsumptionType (%s): %s", req.status_code, data
  128. )
  129. return None
  130. return ConsumptionType.from_dict(data["consumption_type"])
  131. @classmethod
  132. def get(cls, consumption_type_id: int) -> "ConsumptionType":
  133. """ Retrieve a ConsumptionType by id. """
  134. req = requests.get(
  135. urljoin(SERVER_URL, f"/consumption_types/{consumption_type_id}")
  136. )
  137. try:
  138. data = req.json()
  139. if "error" in data:
  140. LOG.warning(
  141. "Could not get consumption type %s (%s): %s",
  142. consumption_type_id,
  143. req.status_code,
  144. data,
  145. )
  146. return None
  147. return cls.from_dict(data["consumption_type"])
  148. except ValueError:
  149. LOG.error(
  150. "Did not get JSON from server on getting consumption type (%s): %s",
  151. req.status_code,
  152. req.content,
  153. )
  154. return None
  155. @classmethod
  156. def get_all(cls) -> ["ConsumptionType"]:
  157. """ Get all active ConsumptionTypes. """
  158. req = requests.get(urljoin(SERVER_URL, "/consumption_types"))
  159. try:
  160. data = req.json()
  161. if "error" in data:
  162. LOG.warning(
  163. "Could not get consumption types (%s): %s", req.status_code, data
  164. )
  165. return [cls.from_dict(item) for item in data["consumption_types"]]
  166. except ValueError:
  167. LOG.error(
  168. "Did not get JSON from server on getting ConsumptionTypes (%s): %s",
  169. req.status_code,
  170. req.content,
  171. )
  172. return None
  173. @classmethod
  174. def from_dict(cls, data: dict) -> "ConsumptionType":
  175. """ Reconstruct a ConsumptionType from a dict. """
  176. return cls(
  177. name=data["name"],
  178. consumption_type_id=data["consumption_type_id"],
  179. icon=data.get("icon"),
  180. )
  181. class Consumption(NamedTuple):
  182. """ Represents a stored Consumption. """
  183. consumption_id: int
  184. person_id: int
  185. consumption_type_id: int
  186. created_at: datetime.datetime
  187. reversed: bool = False
  188. settlement_id: int = None
  189. @classmethod
  190. def from_dict(cls, data: dict) -> "Consumption":
  191. """ Reconstruct a Consumption from a dict. """
  192. return cls(
  193. consumption_id=data["consumption_id"],
  194. person_id=data["person_id"],
  195. consumption_type_id=data["consumption_type_id"],
  196. settlement_id=data["settlement_id"],
  197. created_at=datetime.datetime.fromisoformat(data["created_at"]),
  198. reversed=data["reversed"],
  199. )
  200. def reverse(self) -> "Consumption":
  201. """ Reverse this consumption. """
  202. req = requests.delete(
  203. urljoin(SERVER_URL, f"/consumptions/{self.consumption_id}")
  204. )
  205. try:
  206. data = req.json()
  207. if "error" in data:
  208. LOG.error(
  209. "Could not reverse consumption %s (%s): %s",
  210. self.consumption_id,
  211. req.status_code,
  212. data,
  213. )
  214. return False
  215. return Consumption.from_dict(data["consumption"])
  216. except ValueError:
  217. LOG.error(
  218. "Did not get JSON on reversing Consumption (%s): %s",
  219. req.status_code,
  220. req.content,
  221. )
  222. return False