python-fieldclimate 1.3

事件循环

New in version 1.3.

The same FieldClimateClient class can be used to make asynchronous API requests under any modern event loop. This is thanks to asks being written with anyio, which currently supports asyncio, curio, and trio.

认证

HMAC凭证可以通过几种方式提供

1. 通过init构造函数

   >>> FieldClimateClient(public_key="YOUR", private_key="KEYS")
   

2. 环境变量FIELDCLIMATE_PUBLIC_KEY和FIELDCLIMATE_PRIVATE_KEY。

3. FieldClimateClient子类化

   >>> class MyClient(FieldClimateClient):
   ...     private_key = "YOUR"
   ...     public_key = "KEYS"
   

4. 如果您使用Django，可以使用fieldclimate.django.DjangoFieldClimateClient来代替FieldClimateClient。这个子类将从Django的设置中获取FIELDCLIMATE_PUBLIC_KEY和FIELDCLIMATE_PRIVATE_KEY。

方法

客户端为API文档中列出的每个对应路由都提供了方法。方法很多，因此请参阅fieldclimate/__init__.py中的完整方法列表以获取更多详细信息。每个方法在被等待时都会返回一个类似JSON的Python对象，如字典或列表。

一些方法会清理它们的参数，以便在Python中更容易地使用API。以下是一些示例

• get_data_last()接受time_period参数。API文档指定这应该是一个类似于'6h'或'7d'的字符串，表示6小时或7天。FieldClimateClient还接受timedelta对象作为此参数，并将它们转换为API的等效字符串（即timedelta(hours=6)转换为'21600'秒）。

• 许多方法需要station参数，如上面示例中的get_data_range()所示。这可以是一个原始的Station ID字符串，您可以从get_user_stations()返回的station字典中挖掘出来。或者，您可以直接将那个字典作为station参数传递，并将ID提取出来。

这些方法并不都有测试覆盖率（测试delete_user()可能是个坏主意）。然而，它们使用的底层连接和清理实用工具都已进行了测试。

连接限制

New in version 1.3.

可以通过在调用FieldClimateClient构造函数时设置connections参数来提高连接限制。

请参阅asks的文档

您必须将连接数更改为适合您的需求和服务器限制的值。如果没有公开的数据可以指导您，请保守一些。

会话池中的默认连接数仅为可怜的1个。

示例

async with FieldClimateClient(connections=10) as client:
    ...

根据FieldClimate的文档，他们尚未在服务器端实施速率限制。使用具有高连接限制的FieldClimateClient可以一次创建许多请求。在我的测试中，我发现API开始抛出502错误，当我过度负载它时。

请礼貌地使用您的资源消耗！

高级示例

此函数请求一些用户数据并获取所有用户站点的列表，同时进行。一旦站点返回，它就会计算它们的数量并发送针对前10个站点的另一个请求。然后，按服务器响应时间排序打印这10个站点的每个响应。

from asyncio import gather, run
from fieldclimate import FieldClimateClient

async def main():
    async with FieldClimateClient(
        private_key="YOUR",
        public_key="KEYS",
        connections=20
    ) as client:
        async def print_user_json():
            print(await client.get_user())

        async def print_station_dates(station):
            print(await client.get_data_range(station))

        async def count_stations_then_print_ranges():
            stations = await client.get_user_stations()
            print(len(stations))
            await gather(*[
                print_station_dates(station)
                for station in stations[:10]
            ])

        await gather(
            print_user_json(),
            count_stations_then_print_ranges(),
        )

if __name__ == "__main__":
    run(main())

如果要在curio和trio事件循环中使用FieldClimateClient，请查看tests目录中的替代实现（方法几乎相同）。

同步使用

已从版本1.3中删除。

在版本1.2中，FieldClimateClient会在调用方法时自动设置asyncio事件循环（当方法被调用在async with块之外时）。这样，调用者可以在不编写任何令人害怕的async/await代码的情况下使用库。

这种语法的混合使用变得令人困惑且不必要，同时还导致了这里的代码混乱。因此，随着转向asks后端，对旧同步用例的支持已删除。

如果您使用的是FieldClimateClient的较旧的“同步使用”模式，那么您已经使用了允许async/await的Python版本。区别在于现在您必须自己设置事件循环。

如果你仍然真的不想编写任何协程，让你的代码与1.3版本兼容的最简单方法就是将每个方法调用包裹在asyncio.run()中。

import asyncio
from fieldclimate import FieldClimateClient

def main():
    client = FieldClimateClient(private_key="YOUR", public_key="KEYS")
    # print user json
    print(asyncio.run(client.get_user()))
    # count stations
    stations = asyncio.run(client.get_user_stations())
    print(len(stations))
    # print ranges
    for station in stations[:10]:
        print(asyncio.run(client.get_data_range(station)))

if __name__ == "__main__":
    main()

这个“同步”示例的完成时间比上面的“高级示例”多3倍，因为每次向服务器发送请求时，主()函数都会被阻塞。另一方面，异步代码只有在没有其他事情可做时才会阻塞，只能等待服务器。在决定是否将你的代码转换为使用协程函数时，请考虑这一点。