Public/fastapi-cache
1
0
Fork 0
mirror of https://github.com/long2ice/fastapi-cache.git synced 2026-03-25 04:57:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix #509: fix up linting and tests due to aiobotocore 2.18.0 changes

Browse Source
This commit is contained in:
Gary Gale
2025-01-18 09:52:34 +00:00
parent 7dade61a49
commit 6f4876ff7d
1 changed files with 43 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
fastapi_cache/backends/dynamodb.py
View File

@@ -1,5 +1,5 @@
import datetime import datetime
from typing import TYPE_CHECKING, Optional, Tuple from typing import TYPE_CHECKING, Optional, Tuple, Union
from aiobotocore.client import AioBaseClient from aiobotocore.client import AioBaseClient
from aiobotocore.session import AioSession, get_session from aiobotocore.session import AioSession, get_session
@@ -30,7 +30,7 @@ class DynamoBackend(Backend):
>> FastAPICache.init(dynamodb) >> FastAPICache.init(dynamodb)
""" """
client: DynamoDBClient client: Union[DynamoDBClient, None]
session: AioSession session: AioSession
table_name: str table_name: str
region: Optional[str] region: Optional[str]
@@ -46,9 +46,12 @@ class DynamoBackend(Backend):
).__aenter__() ).__aenter__()
async def close(self) -> None: async def close(self) -> None:
self.client = await self.client.__aexit__(None, None, None) if self.client:
await self.client.__aexit__(None, None, None)
self.client = None
async def get_with_ttl(self, key: str) -> Tuple[int, Optional[bytes]]: async def get_with_ttl(self, key: str) -> Tuple[int, Optional[bytes]]:
if self.client:
response = await self.client.get_item(TableName=self.table_name, Key={"key": {"S": key}}) response = await self.client.get_item(TableName=self.table_name, Key={"key": {"S": key}})
if "Item" in response: if "Item" in response:
@@ -66,12 +69,14 @@ class DynamoBackend(Backend):
return 0, None return 0, None
async def get(self, key: str) -> Optional[bytes]: async def get(self, key: str) -> Optional[bytes]:
if self.client:
response = await self.client.get_item(TableName=self.table_name, Key={"key": {"S": key}}) response = await self.client.get_item(TableName=self.table_name, Key={"key": {"S": key}})
if "Item" in response: if "Item" in response:
return response["Item"].get("value", {}).get("B") return response["Item"].get("value", {}).get("B")
return None return None
async def set(self, key: str, value: bytes, expire: Optional[int] = None) -> None: async def set(self, key: str, value: bytes, expire: Optional[int] = None) -> None:
if self.client:
ttl = ( ttl = (
{ {
"ttl": { "ttl": {