(func, rate="1/5s")
| 170 | # It is up to as the provider to decide how you want |
| 171 | # to rate limit the device during polling. |
| 172 | def rate_limit(func, rate="1/5s"): |
| 173 | def wrapper(): |
| 174 | # some logic to ensure client device is rate limited by a minimum |
| 175 | # of 1 request every 5 seconds during device polling |
| 176 | # https://datatracker.ietf.org/doc/html/rfc8628#section-3.2 |
| 177 | |
| 178 | # use device_code to retrieve device |
| 179 | device = Device |
| 180 | |
| 181 | # get the time in seconds since the device polled the /token endpoint |
| 182 | now = datetime.datetime.now(tz=datetime.UTC) |
| 183 | diff = now - timedelta(device.last_checked) |
| 184 | total_seconds_since_last_device_poll = diff.total_seconds() |
| 185 | |
| 186 | device.last_checked = now |
| 187 | |
| 188 | # for illustrative purposes. 1/5s means 1 request every 5 seconds. |
| 189 | # so if `total_seconds_since_last_device_poll` is 4 seconds, this will |
| 190 | # raise an error |
| 191 | if total_seconds_since_last_device_poll < rate: |
| 192 | raise device_flow_errors.SlowDownError() |
| 193 | |
| 194 | result = func() |
| 195 | return result |
| 196 | |
| 197 | return wrapper |
| 198 | |
| 199 | |
| 200 | class ExampleRequestValidator(RequestValidator): |
nothing calls this directly
no outgoing calls
no test coverage detected
searching dependent graphs…