mirror of
https://github.com/long2ice/fastapi-cache.git
synced 2026-03-25 04:57:54 +00:00
Full mypy --strict type checking pass
This commit is contained in:
@@ -1,11 +1,16 @@
|
||||
import datetime
|
||||
from typing import Optional, Tuple
|
||||
from typing import TYPE_CHECKING, Optional, Tuple
|
||||
|
||||
from aiobotocore.client import AioBaseClient
|
||||
from aiobotocore.session import get_session, AioSession
|
||||
from aiobotocore.session import AioSession, get_session
|
||||
|
||||
from fastapi_cache.backends import Backend
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from types_aiobotocore_dynamodb import DynamoDBClient
|
||||
else:
|
||||
DynamoDBClient = AioBaseClient
|
||||
|
||||
|
||||
class DynamoBackend(Backend):
|
||||
"""
|
||||
@@ -25,9 +30,13 @@ class DynamoBackend(Backend):
|
||||
>> FastAPICache.init(dynamodb)
|
||||
"""
|
||||
|
||||
client: DynamoDBClient
|
||||
session: AioSession
|
||||
table_name: str
|
||||
region: Optional[str]
|
||||
|
||||
def __init__(self, table_name: str, region: Optional[str] = None) -> None:
|
||||
self.session: AioSession = get_session()
|
||||
self.client: Optional[AioBaseClient] = None # Needs async init
|
||||
self.table_name = table_name
|
||||
self.region = region
|
||||
|
||||
@@ -60,6 +69,7 @@ class DynamoBackend(Backend):
|
||||
response = await self.client.get_item(TableName=self.table_name, Key={"key": {"S": key}})
|
||||
if "Item" in response:
|
||||
return response["Item"].get("value", {}).get("B")
|
||||
return None
|
||||
|
||||
async def set(self, key: str, value: bytes, expire: Optional[int] = None) -> None:
|
||||
ttl = (
|
||||
|
||||
Reference in New Issue
Block a user