小编Jam*_*mie的帖子

如何分块加载 Pickle 文件?

是否有任何选项可以分块加载泡菜文件?

我知道我们可以将数据保存在 CSV 中并分块加载。但是除了 CSV 之外,是否有任何选项可以以块的形式加载泡菜文件或任何 Python 本机文件?

python csv file pickle chunks

10
推荐指数
1
解决办法
1750
查看次数

如何在 AWS CDK 创建的 Python Lambda 函数中安装外部模块?

我在 Cloud9 中使用 Python AWS CDK,我正在部署一个简单的 Lambda 函数当对象上传到 S3 存储桶(也由 CDK 创建)时,该函数应该向 Atlassian 的 API发送 API 请求。这是我的 CDK 堆栈代码:

from aws_cdk import core
from aws_cdk import aws_s3
from aws_cdk import aws_lambda
from aws_cdk.aws_lambda_event_sources import S3EventSource


class JiraPythonStack(core.Stack):
    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)

        # The code that defines your stack goes here
        jira_bucket = aws_s3.Bucket(self,
                                    "JiraBucket",
                                    encryption=aws_s3.BucketEncryption.KMS)

        event_lambda = aws_lambda.Function(
            self,
            "JiraFileLambda",
            code=aws_lambda.Code.asset("lambda"),
            handler='JiraFileLambda.handler',
            runtime=aws_lambda.Runtime.PYTHON_3_6,
            function_name="JiraPythonFromCDK") …
Run Code Online (Sandbox Code Playgroud)

python python-module aws-cloudformation aws-lambda aws-cdk

7
推荐指数
3
解决办法
5528
查看次数

如何创建在Python CDK中引用自身的API网关资源策略?

我正在创建一个 API,它接受通过使用具有 GitHub IP 的资源策略从 GitHub Webhook 服务器发出的请求。我已经使用控制台并手动创建资源策略成功完成了此操作,但在使用 CDK 时遇到了问题。

这是我的代码:

delete_trigger_integration = aws_apigateway.LambdaIntegration(
    trigger_step_lambda, proxy=False, integration_responses=[])

api_policy_document = aws_iam.PolicyDocument()

api = aws_apigateway.RestApi(
    self,
    "GithubWebhookApi",
    rest_api_name=PROJECT_NAME + "-apigateway-trigger-delete",
    default_integration=delete_trigger_integration,
    policy=api_policy_document)

delete_execution_resource = api.root.add_resource("execution")

delete_execution_method = delete_execution_resource.add_method(
    "POST", delete_trigger_integration)
delete_execution_resource.add_cors_preflight(allow_origins=["*"])

create_repo_lambda.add_environment("API_URL",
                                   delete_execution_resource.url)

api_policy_document.add_statements(
    aws_iam.PolicyStatement(
        effect=aws_iam.Effect.ALLOW,
        principals=[aws_iam.AnyPrincipal()],
        actions=["execute-api:Invoke"],
        resources=[api.arn_for_execute_api()]))

api_policy_document.add_statements(
    aws_iam.PolicyStatement(
        effect=aws_iam.Effect.DENY,
        actions=["execute-api:Invoke"],
        conditions={
            "NotIpAddress": {
                "aws:SourceIp": [
                    "192.30.252.0/22", "185.199.108.0/22",
                    "140.82.112.0/20"
                ]
            }
        },
        principals=[aws_iam.AnyPrincipal()],
        resources=[api.arn_for_execute_api()]))
Run Code Online (Sandbox Code Playgroud)

我觉得我已经非常接近找到解决方案,但我不知道它是什么。上面代码的问题是我ValidationError: Circular dependency between resources在尝试部署它时收到错误 - 我可以理解,资源策略正在解决它内部的资源。但我在 CDK 中找不到解决方法。

对于其他资源,在创建后使用 …

python aws-cloudformation amazon-iam aws-api-gateway aws-cdk

7
推荐指数
1
解决办法
6410
查看次数

如何在 JavaScript 中使用 CDK 向现有 S3 存储桶添加资源策略?

我正在将资源策略文档添加到 S3 存储桶。

当我创建一个新的 Bucket 时它工作正常:

const newbucket = new s3.Bucket(this, 'newBucket', {
      websiteIndexDocument : 'index.html',
      bucketName : 'NewBucket'
});

newbucket.addToResourcePolicy(new iam.PolicyStatement({
      effect : iam.Effect.ALLOW,
      actions: ['s3:*'],
      resources: [newbucket.arnForObjects('*')],
      principals: [new iam.AnyPrincipal],
    }));

newbucket.addToResourcePolicy(new iam.PolicyStatement({
      effect : iam.Effect.DENY,
      actions: ['s3:*'],
      resources: [newbucket.arnForObjects('*')],
      principals: [new iam.AnyPrincipal],
      conditions : {
        'NotIpAddress' : {
          'aws:SourceIp' : '***.***.***.***'
        }
      }
    }));
Run Code Online (Sandbox Code Playgroud)

但是,如果我尝试获取已经存在的存储桶并添加策略文档,则它不起作用:

const existingbucket = Bucket.fromBucketAttributes(this, 'ImportedBucket',{
      bucketName :'ExistingBucket'
    })

existingbucket.addToResourcePolicy(new iam.PolicyStatement({
      effect : iam.Effect.ALLOW,
      actions: ['s3:*'],
      resources: [existingbucket.arnForObjects('*')],
      principals: [new iam.AnyPrincipal],
    }));
Run Code Online (Sandbox Code Playgroud)

不会添加资源策略文档。 …

javascript amazon-s3 aws-cdk

7
推荐指数
1
解决办法
2416
查看次数

尝试添加分区时 Athena 查询“输入‘分区’不匹配”

这是我的雅典娜查询。我正在尝试向表添加分区:

ALTER TABLE public_data_scraping_yahoo_finance_pricing_table
ADD PARTITION ("S3_DATE" = '2021-07') 
LOCATION 's3://my-bucket/enriched_data/pricing/2021-07/';
Run Code Online (Sandbox Code Playgroud)

我正在遵循此处所见的文档。但我遇到了某种语法错误。

当我运行它时,我得到:

line 2:5: mismatched input 'partition'. expecting: '.', 'add' (service: amazonathena; status code: 400; error code: invalidrequestexception; request id: 8dd2f1ac-d197-4c4c-b2fc-024383fd30fb; proxy: null)
Run Code Online (Sandbox Code Playgroud)

我之前曾运行过其他查询来添加分区,但从未遇到过此问题。我希望这个查询能够毫无问题地创建分区。

有人能够帮助我确定问题吗?

sql amazon-athena

7
推荐指数
2
解决办法
1万
查看次数

如何定期运行CDK中创建的Lambda?

正如标题所说 - 我在 Python CDK 中创建了一个 Lambda,我想知道如何定期触发它(例如每天一次)。

我确信这是可能的,但我是 CDK 的新手,我正在努力寻找解决文档的方法。据我所知,它将使用某种事件触发器 - 但我不确定如何使用它。

任何人都可以帮忙吗?

amazon-web-services aws-cdk

6
推荐指数
1
解决办法
4727
查看次数

Pytest 的“caplog”装置的类型提示是什么?

正如标题所说。我正在使用caplogpytest 附带的夹具。我正在使用 mypy 进行类型检查,并且想知道 caplog 的正确类型提示是什么。

例如:

def test_validate_regs(caplog: Any) -> None:
    validate_regs(df, logger)
    assert caplog.text == "", "No logs should have been made."
Run Code Online (Sandbox Code Playgroud)

在此示例中,我将其设置为Any,但我想知道是否可以使用更具体的类型提示。

我尝试阅读有关 caplog 的文档,以及在 github 中搜索 pytest 代码以查看 caplog 固定装置返回的内容,但除了以下内容之外找不到任何其他内容。但是使用str类型只是给了我一个错误,说该str类型没有属性text,这是有道理的。

当我打印 caplog 的类型时,_pytest.logging.LogCaptureFixture虽然我不确定如何从中导入_pytest并使用它。

python pytest mypy

2
推荐指数
1
解决办法
109
查看次数

我正在努力在多个函数中使用变量(赋值之前引用了变量!)

我是Python的新手,真的很喜欢代码。我正在尝试重新创建一个我的朋友在Python中制作的人口模拟器。这真的很基础,目前仅使用3个变量。这是我的代码:

pop = 7000000000
int = 5

#Clear
def cls():
    print "\n" * 80

#Main script
def MS():
    cls()
    print "Population =", pop, "Intelligence =", int
    print ""
    print "Do You:"
    print "A) Increase Population"
    print "B) Increase Intelligence"
    choice = raw_input("Option:")
    if choice == "a" :
        print "option a"
    elif choice == "A" :
        print "option A"
    elif choice == "b" :
        intel()
    elif choice == "B" :
        intel()

#Increase Intelligence
def intel():
    int = int + 3

MS() …
Run Code Online (Sandbox Code Playgroud)

python variables python-2.7

1
推荐指数
1
解决办法
2304
查看次数