【问题标题】:Making an API request (LinkedIn) for every element in a list from a response, with Python使用 Python 为响应列表中的每个元素发出 API 请求 (LinkedIn)
【发布时间】:2021-12-04 00:14:15
【问题描述】:

我有一个 LinkedIn 帖子 ID 列表。我需要通过另一个请求请求每个帖子的共享统计信息。

请求函数如下所示:

def ugcp_stats(headers):
    response = requests.get(f'https://api.linkedin.com/v2/organizationalEntityShareStatistics?q=organizationalEntity&organizationalEntity=urn%3Ali%3Aorganization%3A77487&ugcPosts=List(urn%3Ali%3AugcPost%3A{shid},urn%3Ali%3AugcPost%3A{shid2},...,urn%3Ali%3AugcPost%3A{shidx})', headers = headers)
    ugcp_stats = response.json()
    return ugcp_stats

urn%3Ali%3AugcPost%3A{shid},urn%3Ali%3AugcPost%3A{shid2},...,urn%3Ali%3AugcPost%3A{shidx} - 这些是共享 urn。它们的数量取决于我列表中的元素数量。

接下来我该怎么做?我是否应该计算列表中的元素数量并以某种方式修改请求 URL 以包含所有元素?或者也许我应该遍历列表并对每个元素发出单独的请求,然后将所有响应附加到一个 json 文件中?

我很挣扎,我不太确定如何写这个。我什至不知道如何将元素解析为请求。虽然我怀疑它可能看起来像这样:

for shid in shids:
    def ugcp_stats(headers):
        response = requests.get(f'https://api.linkedin.com/v2/organizationalEntityShareStatistics?q=organizationalEntity&organizationalEntity=urn%3Ali%3Aorganization%3A77487&ugcPosts=List(urn%3Ali%3AugcPost%3A & {shid})', headers = headers)
        ugcp_stats = response.json()
        return ugcp_stats

更新 - 关注您的分析 现在的代码如下所示:

    link = "https://api.linkedin.com/v2/organizationalEntityShareStatistics?q=organizationalEntity&organizationalEntity=urn%3Ali%3Aorganization%3A77487&ugcPosts=List"
def share_stats(headers, shids):
    # Local variable
    sample = ""
    # Sample the shids in the right pattern
    for shid in shids: sample += "urn%3Ali%3AugcPost%3A & {},".format(shid)
    # Get the execution of the string content
    response = eval(f"requests.get('{link}({sample[:-1]})', headers = {headers})")
    # Return the stats
    return response.json()
    if __name__ == '__main__':
        credentials = 'credentials.json'
        access_token = auth(credentials) # Authenticate the API
        headers = headers(access_token) # Make the headers to attach to the API call.
        share_stats = share_stats(headers) # Get shares
    print(share_stats)

但似乎什么都没有发生。它完成了脚本,但我什么也没得到。怎么了?

【问题讨论】:

  • 如果您确实发现需要为此用例发出多个 API 请求,我建议使用 concurrent.futures.ThreadPoolExecutor 并且可能使用它的“地图”函数 IIRC提供 - 感觉它很适合这个工作流程。
  • 您可以使用 exec 或 eval。他们做同样的事情,两者都会从字符串中执行 python 代码。因此会将您的消息解释为 python 代码。他们之间的区别正在回归。 eval 也将返回执行的值,而 exec 将只执行而不返回值(如果它)。这将帮助您自动化链接字符串类型中的“shid”提要。还可以考虑将@rv.kvetch 建议作为一种优化策略,以便从现在开始利用多线程技术。

标签: python python-3.x linkedin-api


【解决方案1】:

这只是我之前作为评论告诉你的证明。现在您将适应您的需求(即使我尝试为您做):)

更新 - 根据您的反馈。

#// IMPORT
#// I'm assuming your are using "requests" library
#// PyCharm IDE show me like this library is not used, but "eval()" is using it
import requests

#// GLOBAL VARIABLES
link: str = "https://api.linkedin.com/v2/organizationalEntityShareStatistics?q=organizationalEntity&organizationalEntity=urn%3Ali%3Aorganization%3A77487&ugcPosts=List"


#// Your function logic updated
def share_stats(sheds: list, header: dict) -> any:
    # Local variable
    sample = ""

    # Sample the sheds in the right pattern
    for shed in sheds: sample += "urn%3Ali%3AugcPost%3A & {},".format(shed)

    # Get the execution of the string content
    response = eval(f"requests.get('{link}({sample[:-1]})', headers = {header})")

    # Return the stats as JSON file
    return response.json()


#// Run if this is tha main file
if __name__ == '__main__':
    #// An empty sheds list for code validation
    debug_sheds: list = []
    credentials: str = "credentials.json"

    #// For me I get an unresolved reference for "auth", for you shod be fine
    #// I'm assuming is your function for reading the file content and convert it to Python
    access_token = auth(credentials)    # Authenticate the API

    #// Your error was from this script line
    #// Error message: 'TypedDict' object is not callable
    #// Your code: headers = headers(access_token)
    #// When you want to get a dictionary value by a key use square brackets
    headers = headers[access_token]     # Make the headers to attach to the API call.

    #// Here you shood ged an error/warning because you do not provided the sheds first time
    #// Your code: share_stats = share_stats(headers)
    share_stats = share_stats(debug_sheds, headers)  # Get shares

    print(share_stats)

【讨论】:

  • 感谢您的回复!我正在尝试使您的功能正常工作,但出现类型错误。我修改了第一篇文章以显示代码现在的样子。 ``` headers = headers(access_token) # 使标头附加到 API 调用。 TypeError:'dict'对象不可调用```
  • @PiotrChmurzynski if __name__ == '__main__': 需要在函数定义之外。如果主脚本文件是这个脚本,那部分脚本的短篇故事是运行,其他方式文件的洞内容可以用作导入类和函数的库。有关更多信息,您可以阅读this answer for the forum。试试看,让我知道它正在工作。我仍然对你的字典的样子感到困惑。
  • @PiotrChmurzynski 或者至少我是这样使用它的。此外,我将您的代码复制粘贴到 PyCharm 上,并且 IDE 在返回状态后播种该脚本无法访问。您可能想要修改那个。还要检查this forum page 以查看您的字典调用是否使用错误或更改headers 变量名,可能是headers = {headers}) 上的类型名冲突。其他方式更新和“标题”值来制作和了解您的字典的外观(例如print(headers(auth('credentials.json'))))。
猜你喜欢
  • 1970-01-01
  • 2021-04-21
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多