Update protocol ST refresh configuration.
(
self,
enabled: Optional[bool] = None,
refresh_interval_minutes: Optional[int] = None,
)
| 2028 | return TokenRefreshConfig() |
| 2029 | |
| 2030 | async def update_token_refresh_config( |
| 2031 | self, |
| 2032 | enabled: Optional[bool] = None, |
| 2033 | refresh_interval_minutes: Optional[int] = None, |
| 2034 | ) -> TokenRefreshConfig: |
| 2035 | """Update protocol ST refresh configuration.""" |
| 2036 | async with self._connect(write=True) as db: |
| 2037 | db.row_factory = aiosqlite.Row |
| 2038 | cursor = await db.execute("SELECT * FROM token_refresh_config WHERE id = 1") |
| 2039 | row = await cursor.fetchone() |
| 2040 | current = dict(row) if row else {} |
| 2041 | new_enabled = enabled if enabled is not None else bool(current.get("enabled", True)) |
| 2042 | try: |
| 2043 | new_interval = int( |
| 2044 | refresh_interval_minutes |
| 2045 | if refresh_interval_minutes is not None |
| 2046 | else current.get("refresh_interval_minutes", 120) |
| 2047 | ) |
| 2048 | except Exception: |
| 2049 | new_interval = 120 |
| 2050 | new_interval = max(1, new_interval) |
| 2051 | |
| 2052 | if row: |
| 2053 | await db.execute(""" |
| 2054 | UPDATE token_refresh_config |
| 2055 | SET enabled = ?, refresh_interval_minutes = ?, updated_at = CURRENT_TIMESTAMP |
| 2056 | WHERE id = 1 |
| 2057 | """, (new_enabled, new_interval)) |
| 2058 | else: |
| 2059 | await db.execute(""" |
| 2060 | INSERT INTO token_refresh_config (id, enabled, refresh_interval_minutes) |
| 2061 | VALUES (1, ?, ?) |
| 2062 | """, (new_enabled, new_interval)) |
| 2063 | await db.commit() |
| 2064 | |
| 2065 | return await self.get_token_refresh_config() |
no test coverage detected