doorcontrol: Improve pagination behavior of DoorController.get_records()
Use `DR` method to get total count of elements then paginate by defined page size, instead of hacky bad automatically sized pagination
This commit is contained in:
parent
017e70b7d1
commit
32a91315ef
@ -2,6 +2,7 @@ import contextlib
|
||||
import csv
|
||||
from datetime import datetime
|
||||
from io import StringIO
|
||||
from itertools import takewhile
|
||||
|
||||
import requests
|
||||
import urllib3
|
||||
@ -33,6 +34,14 @@ class RemoteError(Exception):
|
||||
super().__init__(f"Door Updating Error: {r.status_code} {r.reason}\n{r.text}")
|
||||
|
||||
|
||||
class UnsupportedPageSize(Exception):
|
||||
def __init__(self, page_size: int) -> None:
|
||||
super().__init__(
|
||||
f"Page size {page_size} greater than supported by controller. "
|
||||
"(controller returned moreRecords=true)"
|
||||
)
|
||||
|
||||
|
||||
class DoorController:
|
||||
def __init__(self, ip, username, password):
|
||||
self.ip = ip
|
||||
@ -152,48 +161,44 @@ class DoorController:
|
||||
)
|
||||
return self.doXMLRequest(el)
|
||||
|
||||
def get_records(self, req, count, params=None, stopFunction=None):
|
||||
recordCount = 0
|
||||
moreRecords = True
|
||||
def get_records(
|
||||
self,
|
||||
req,
|
||||
count_attr: str,
|
||||
params: dict[str, str] | None = None,
|
||||
page_size: int = 100,
|
||||
):
|
||||
dr = self.doXMLRequest(ROOT(req({"action": "DR"})))
|
||||
|
||||
# note: all the "+/-1" bits are to work around a bug where the
|
||||
# last returned entry is incomplete. There is probably a
|
||||
# better way to do this, but for now I just get the last entry
|
||||
# again in the next request. I suspect this probably ends
|
||||
# poorly if the numbers line up poorly (ie an exact multiple
|
||||
# of the returned record limit)
|
||||
while True:
|
||||
for offset in range(0, int(dr[0].attrib[count_attr]), page_size):
|
||||
res = self.doXMLRequest(
|
||||
ROOT(
|
||||
req(
|
||||
{
|
||||
"action": "LR",
|
||||
"recordCount": str(count - recordCount + 1),
|
||||
"recordOffset": str(
|
||||
recordCount - 1 if recordCount > 0 else 0
|
||||
),
|
||||
"recordCount": str(page_size),
|
||||
"recordOffset": str(offset),
|
||||
**(params or {}),
|
||||
}
|
||||
)
|
||||
)
|
||||
)
|
||||
recordCount += int(res[0].get("recordCount")) - 1
|
||||
moreRecords = res[0].get("moreRecords") == "true"
|
||||
|
||||
if moreRecords and (stopFunction is None or stopFunction(list(res[0]))):
|
||||
yield list(res[0])[:-1]
|
||||
else:
|
||||
yield list(res[0])
|
||||
break
|
||||
# The web interface does sub-pagination when needed, but that is very messy.
|
||||
# See previous versions of this function for an example :)
|
||||
if res[0].attrib["moreRecords"] != "false":
|
||||
raise UnsupportedPageSize(page_size)
|
||||
|
||||
yield list(res[0])
|
||||
|
||||
def get_cardholders(self):
|
||||
for page in self.get_records(
|
||||
E.Cardholders, 1000, {"responseFormat": "expanded"}
|
||||
E.Cardholders, "cardholdersInUse", params={"responseFormat": "expanded"}
|
||||
):
|
||||
yield from page
|
||||
|
||||
def get_credentials(self):
|
||||
for page in self.get_records(E.Credentials, 1000):
|
||||
for page in self.get_records(E.Credentials, "credentialsInUse"):
|
||||
yield from page
|
||||
|
||||
def update_credential(self, rawCardNumber: str, cardholderID: str):
|
||||
@ -210,19 +215,17 @@ class DoorController:
|
||||
)
|
||||
)
|
||||
|
||||
def get_events(self, threshold):
|
||||
def get_events(self, threshold: datetime):
|
||||
def event_newer_than_threshold(event):
|
||||
return datetime.fromisoformat(event.attrib["timestamp"]) > threshold
|
||||
|
||||
# These door controllers only store 5000 events max
|
||||
for page in self.get_records(
|
||||
E.EventMessages,
|
||||
5000,
|
||||
stopFunction=lambda events: event_newer_than_threshold(events[-1]),
|
||||
):
|
||||
events = [event for event in page if event_newer_than_threshold(event)]
|
||||
# smaller page size empirically determined
|
||||
for page in self.get_records(E.EventMessages, "eventsInUse", page_size=25):
|
||||
events = list(takewhile(event_newer_than_threshold, page))
|
||||
if events:
|
||||
yield events
|
||||
else:
|
||||
break
|
||||
|
||||
def get_lock(self):
|
||||
el = ROOT(E.Doors({"action": "LR", "responseFormat": "status"}))
|
||||
|
Loading…
Reference in New Issue
Block a user