如何使用python根据特定时间范围过滤API响应数据

Question

我使用一个lambda python函数来使用mailgun日志API从mailgun获取电子邮件日志。这是我的功能，

import json
import requests
resp = requests.get("https://api.eu.mailgun.net/v3/domain/events",
                    auth=("api","key-api"))
def jprint(obj):
    # create a formatted string of the Python JSON object
    text = json.dumps(obj, sort_keys=True, indent=4)
    print(text)
jprint(resp.json())

此函数提供从mailgun API获取的电子邮件日志的格式化json输出。

来自API的示例响应，

{
    "items": [
        {
            "campaigns": [],
            "delivery-status": {
                "attempt-no": 1,
                "certificate-verified": true,
                "code": 250,
                "description": "",
                "message": "OK",
                "mx-host": "host",
                "session-seconds": 1.5093050003051758,
                "tls": true
            },
            "envelope": {
                "sender": "postmaster@domain.com",
                "sending-ip": "ip",
                "targets": "id@mail.com",
                "transport": "smtp"
            },
            "event": "delivered",
            "flags": {
                "is-authenticated": true,
                "is-routed": false,
                "is-system-test": false,
                "is-test-mode": false
            },
            "id": "id",
            "log-level": "info",
            "message": {
                "attachments": [],
                "headers": {
                    "from": "NAME <noreply@name.com>",
                    "message-id": "20220223075827.de300265fad746e9@domain.com",
                    "subject": "Client due diligence information has been submitted by one of your customers.",
                    "to": "id@mail.com"
                },
                "size": 1990
            },
            "recipient": "id@mail.com",
            "recipient-domain": "domain.com",
            "storage": {
                "key": "key",
                "url": "https://storage.eu.mailgun.net/v3/domains/domain/messages/id"
            },
            "tags": [],
            "timestamp": 1645603109.434181,
            "user-variables": {}
        },
        {
            "envelope": {
                "sender": "postmaster@domain.com",
                "targets": "id@mail.com.com",
                "transport": "smtp"
            },
            "event": "accepted",
            "flags": {
                "is-authenticated": true,
                "is-test-mode": false
            },
            "id": "id",
            "log-level": "info",
            "message": {
                "headers": {
                    "from": "NAME <noreply@name.com>",
                    "message-id": "20220223075827.de300265fad746e9@domain.com",
                    "subject": "Client due diligence information has been submitted by one of your customers.",
                    "to": "id@mail.com.com"
                },
                "size": 1990
            },
            "method": "HTTP",
            "recipient": "id@mail.com",
            "recipient-domain": "domain",
            "storage": {
                "key": "key",
                "url": "https://storage.eu.mailgun.net/v3/domains/domain/messages/key"
            },
            "tags": null,
            "timestamp": 1645603107.282775,
            "user-variables": {}
        },

在这里，时间戳不是人类可读的。

我需要设置aws lambda python脚本来触发事件，定期调用mailgun并将日志发送到cloudwatch。我熟悉设置，但不熟悉脚本。

现在，我只需要在最后一个小时内动态地过滤API数据。

通过使用熊猫库的分析，这是可以实现的，但我无法得到正确的答案来定期获取动态时间范围内的日志。

关于这一点，我参考了很多文档，但是我找不到合适的答案，而且python对我来说也是全新的。

有人能指点我如何动态地从最后的N个时间范围内得到日志吗？

Answer 1

除了上述答案之外，对于仅使用python的两个时间范围和日期范围之间的筛选，您可以使用datetime。这里使用与@Amirhossein Kiani相同的列表理解。

import datetime

start = datetime.datetime(year, month, day, hour, minute, second).timestamp()
stop = datetime.datetime(year, month, day, hour, minute, second).timestamp()

[x for x in data["items"] if start < x["timestamp"] < stop]

对于一个小时的差异，您还可以使用timedelta：

start = (datetime.datetime.now() - datetime.timedelta(hours=1)).timestamp()
stop = datetime.datetime.now().timestamp()

Answer 2

在梅尔贡的文档中，您可以指定一个定时器，这样就可以使用begin和end 参数过滤结果。

在此之后，您可以使用pd.json_normalize重塑您的json响应。

Answer 3

除了我个人更喜欢的关于文档的@Corr外星人所说的内容之外，您还可以使用纯python方法使用列表理解来重新获取最后一个小时的数据。在下面的代码中，我将假设您将API的响应命名为data，应该是字典：

from time import time
lastHour = time() - 3600
[x for x in data["items"] if x["timestamp"] > lastHour]

这将使用大于最后一个小时(time() - 3600)的时间戳过滤值。