dynamo.py 文件源码

def get_last_updated(self, source_id):
        params = {
            "TableName": self.table_name,
            "IndexName": "source_id-updated-index",
            "KeyConditionExpression": "source_id = :value",
            "ExpressionAttributeValues": {
                ":value": {"S": str(source_id)},
            },
            "ProjectionExpression": "source_id, updated, post_id",
            "ScanIndexForward": False,
            "Limit": 1,
        }
        db = await self.db
        try:
            response = await db.query(**params)
            if response.get("ResponseMetadata", {}).get("HTTPStatusCode") == 200:
                updated = response["Items"][0]["updated"]["S"] if response["Count"] == 1 else None
                tstamp = parse_date(updated) if updated else None
                return tstamp
        except BotoCoreError as exc:
            logger.error("[DB] Error when querying for last updated item on {}".format(source_id))
            logger.error(exc)
        return None