小编Jak*_*ake的帖子

我想使用 pydantic 进行模式验证，并使用 TypedDict 来定义嵌套字典模式的一部分。但是，我意识到Optional如果在 TypedDict 类中指定它则不起作用。

我读到这个类将根据需要呈现所有键，并且使所有键都成为可选的方法是 at total=False。但是，我只希望其中一个键是可选的，其余键是必需的。有没有办法克服这个限制？

from typing import List, Optional
from pydantic import BaseModel
from typing_extensions import TypedDict

class _trending(TypedDict):
    allStores: Optional[bool] = False
    category: str
    date: str
    average: List[int]

class RequestSchema(BaseModel):
    storeId: str
    trending: _trending

编辑

我之前尝试过这个，因为我认为它类似于嵌套列表。

from typing import List, Optional, Dict
from pydantic import BaseModel


class _trending(BaseModel):
    allStores: Optional[bool] = False
    category: str
    date: str
    average: List[int]

class RequestSchema(BaseModel):
    storeId: str
    trending: Dict[_trending]

但遇到一条错误消息，指出 Dict 需要 2 个参数。显然 Dict …

我试图在 CI/CD 管道中承担 AWS 角色，因此我必须编写一个脚本来通过脚本更改角色。下面是执行此操作的脚本，我用来source <script>.sh替换现有的 AWS 访问和密钥，并添加会话密钥。

我通过在终端中回显 3 个环境变量来检查它们是否存在。

#!/bin/bash

output="/tmp/assume-role-output.json"

aws sts assume-role --role-arn "arn:aws:iam::<account-id>:role/<rolename>" --role-session-name AWSCLI-Session > $output
AccessKeyId=$(cat $output | jq '.Credentials''.AccessKeyId')
SecretAccessKey=$(cat $output | jq '.Credentials''.SecretAccessKey')
SessionToken=$(cat $output | jq '.Credentials''.SessionToken')

export AWS_ACCESS_KEY_ID=$AccessKeyId
export AWS_SECRET_ACCESS_KEY=$SecretAccessKey
export AWS_SESSION_TOKEN=$SessionToken

但是，当我尝试运行简单的 aws 命令来列出 ECR 映像时aws ecr list-images --registry-id <id> --repository-name <name>，它给出了以下错误消息。

An error occurred (UnrecognizedClientException) when calling the ListImages operation: 
The security token included in the request is invalid.

我尝试在终端中手动设置 AWS 密钥和令牌，令人惊讶的是 ecr …

有许多类似的问题,但没有具体到这个问题.

我有一个数据框列表,我需要使用一个唯一的列将它们合并在一起(date).字段名称不同,因此concat已经出局.

我可以手动使用df[0].merge(df[1],on='Date').merge(df[3],on='Date)等由一个合并的每个DF之一,但问题是,数据帧中的列表中的号码与用户输入不同.

有没有什么方法可以合并,只是一次性组合列表中的所有数据帧？或者也许有些人会在循环中做到这一点？

我使用的是Python 2.7.

这是我第一次尝试生成 gitlab-ci 作业工件。我的 ci 脚本生成一个 csv 文件，我想将其下载为作业中的工件。yaml 文件如下。

   unittest:
      script:
        - cd unittest
        - bash ci-test.sh
      artifacts:
        paths:
          - /*.csv
        when: always
        expire_in: 1 day

我收到了这样的错误消息。

Uploading artifacts for successful job
00:02
Uploading artifacts...
WARNING: /*.csv: no matching files 
ERROR: No files to upload 

我已经确认生成的 csv 报告在那里。

我是 D3 的初学者，并试图掌握如何在 D3 中构建适当的力导向图的诀窍。似乎有两种不同的构建方式。

1. 使用 layout.force。例如
2. 使用 forceSimulation。例如

我或多或少对第一个感到满意，因为代码更短且更容易理解。但是，如果没有正确理解，很难将使用示例 2 的示例代码中的函数转换为示例 1。任何人都能够描述每个的优点或缺点吗？

谢谢你。

我正在尝试使用从另一个网站下载的 pip 将以下安装命令转换为 requirements.txt 格式，但无法弄清楚如何。任何人都可以提供帮助吗？

pip install torch==1.5.0+cu101 torchvision==0.6.0+cu101 -f https://download.pytorch.org/whl/torch_stable.html

pip install detectron2 -f https://dl.fbaipublicfiles.com/detectron2/wheels/cu101/index.html

我是新手来安装新的python模块.

我使用pip install tweepy安装了tweepy.安装成功,在Lib/site-packages中创建了2个文件夹tweepy和tweepy-3.3.0.dist-info,因此我认为一切都应该没问题.

但是,当我去IDE并导入tweepy时.它无法检测到模块:

>>> import tweepy
Traceback (most recent call last):
  File "<interactive input>", line 1, in <module>
ImportError: No module named tweepy

怎么了？

我正在运行python 2.7.5.

[更新1]我正在使用Windows 7.

我首先使用另一个论坛的建议安装了pip(如何在Windows上安装pip？).基本上保存get-pip.py脚本并双击它(无法让"python get-pip.py"按照建议在cmd提示符下工作).然后,我去了cmd并且唠叨到C:/ Python27/Scripts并输入pip install tweepy.我记得看到结果是一个成功的安装.

[更新2]使用带有导入tweepy的文件并运行它,我有类似的错误.

Traceback (most recent call last):
  File "C:\Python27\ArcGIS10.2\Lib\site-packages\pythonwin\pywin\framework\scriptutils.py", line 326, in RunScript
    exec codeObject in __main__.__dict__
  File "C:\Users\xxxx\Desktop\Script1.py", line 2, in <module>
    from tweepy import Stream
ImportError: No module named tweepy

[更新3]在cmd中键入"pip freeze".它确实显示tweepy = 3.3.0

C:\Python27\Scripts>pip freeze 
oauthlib==0.7.2 …

我正在尝试使用 spark 结构化流 (spark v2.2.0) 来使用来自 kafka 的 json 数据。但是我遇到了以下错误。

pyspark.sql.utils.StreamingQueryException: '缺少没有默认值的必需配置“partition.assignment.strategy”。

有谁知道为什么？该作业是使用下面的 spark-submit 提交的。

spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0 sparksstream.py

这是整个python脚本。

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

spark = SparkSession \
    .builder \
    .appName("test") \
    .getOrCreate()

# Define schema of json
schema = StructType() \
        .add("Session-Id", StringType()) \
        .add("TransactionTimestamp", IntegerType()) \
        .add("User-Name", StringType()) \
        .add("ID", StringType()) \
        .add("Timestamp", IntegerType())

# load data into spark-structured streaming
df = spark \
      .readStream \
      .format("kafka") \
      .option("kafka.bootstrap.servers", "xxxx:9092") …

我必须提供一个 API，该 API 向其他 API 发送多个请求并将它们的响应合并为输出。

做到这一点的最佳方法是什么？这是某种异步等待场景吗？感谢您的建议。

from fastapi import FastAPI
import requests

app = FastAPI()

@app.post('/api')
def main_api():
    JScontent = json.loads(request.json())
    input = JScontent['content']
    response1 = requests.post(url1, json={"input":input})
    response2 = requests.post(url2, json={"input":input})
    response3 = requests.post(url3, json={"input":input})
    response4 = requests.post(url4, json={"input":input})

    prediction = fuse_responses(response1, response2, response3, response4)
    return prediction

我目前正在使用 Flask 进行开发，但认为它可能没有能力，或者管理这样的场景会很麻烦，因此，愿意更改为 FastAPI。

[更新]

是否有一些挖掘发现了这个链接。我想它可以像FastAPI一样应用吗？我之前没有异步和等待方面的经验，因此感谢验证。

from fastapi import FastAPI


app = FastAPI()

async def send_requests(url, input_):
    res = await app.post(url, input_)
    return res
    

@app.post('/api')
async def main_api():
    JScontent …

我正在尝试使用 Selenium 来（1）在网站中提交查询，然后（2）使用 beautiful soup 复制结果的内容。这是我第一部分的脚本......

from selenium import webdriver

browser = webdriver.Chrome('C:\Users\XXX\Scripts\MyPythonScripts\chromedriver.exe')
browser.get(r'http://www.ars-grin.gov/cgi-bin/npgs/html/tax_search.pl?language=en')

elem = browser.find_element_by_name('search')
elem.send_keys('Syzygium polyanthum')
elem.submit()

对于第二部分，我意识到我必须以某种方式将结果的新 url 复制到变量中，然后才能使用 beautiful soup 来获取内容，但在广泛搜索后我不知道如何做到这一点。

有谁知道这一点，或者有任何替代方法可以达到相同的结果？