小编Pau*_*gel的帖子

这是我一直想知道的事情.我将提出Python的问题,但我也欢迎解决Java和C++标准库的问题.

假设您有一个名为"my_list"的Python列表,并且您希望迭代其独特元素.有两种自然方法:

#iterate over set
for x in set(my_list):
    do_something(x)

要么

#list to set to list
for x in list(set(my_list)):
    do_something(x)

紧张是在列表上迭代比在集合上迭代更快,但是将集合转换为列表需要时间.我的猜测是,这个问题的答案取决于许多因素,例如:

• 我们需要迭代多少次？
• 原始清单有多大？
• 我们应该在原始列表中重复多少次？

所以我想我正在寻找形式的经验法则"如果列表有x个元素,每个元素重复不超过y次,你只需要迭代z次,那么你应该遍历集合;否则你应该把它转换成一个清单."

该软件包boto3- 亚马逊用于python的官方AWS API包装器 - 非常支持批量上传项目到DynamoDB.它看起来像这样:

db = boto3.resource("dynamodb", region_name = "my_region").Table("my_table")

with db.batch_writer() as batch:
    for item in my_items:
        batch.put_item(Item = item)

这my_items是一个Python字典列表,每个字典都必须包含表的主键.情况并不完美 - 例如,没有安全机制来阻止您超出吞吐量限制 - 但它仍然相当不错.

但是,从数据库中读取似乎没有任何对应物.我能找到的最接近的是DynamoDB.Client.batch_get_item(),但这里的API非常复杂.这是请求两个项目的样子:

db_client = boto3.client("dynamodb", "my_region")

db_client.batch_get_item(
    RequestItems = {
        "my_table": {
            "Keys": [
                {"my_primary_key": {"S": "my_key1"}},
                {"my_primary_key": {"S": "my_key2"}}
            ]
        }
    }
)

这可能是可以容忍的,但是响应也存在同样的问题:所有值都是字典,其键是数据类型("S"对于字符串,"N"数字,"M"映射等),并且解析所有内容会有点烦人.所以我的问题是:

是否有任何原生boto3支持从DynamoDb批量读取,类似于batch_writer上面的功能？

没错,

是否boto3提供任何内置方法来自动反序列化对DynamoDB.Client.batch_get_item()函数的响应？

我还要补充说,该函数boto3.resource("dynamodb").Table().get_item()具有我认为的"正确"API,因为输入或输出不需要类型解析.所以这似乎是开发人员的某种疏忽,我想我正在寻找一种解决方法.

以下是 python 中抽象基类的典型用法：

from abc import ABC

class MyBaseClass(ABC):
    @abstractmethod
    def __init__(self):
        print("Here's some logic that I want all subclasses to have")

class ChildClass(MyBaseClass):
    def __init__(self):
        super().__init__()
        print("Here's some additional logic that I want this subclass to have")

行为是：不可能实例化MyBaseClass，但可以对其进行子类化，并MyBaseClass包括所有子类继承的逻辑。

这是获得相同行为的另一种方法：

class MyBaseClass:
    def __init__(self):
        if type(self) == MyBaseClass:
            raise NotImplementedError
        print("Here's some logic that I want all subclasses to have")

class ChildClass(MyBaseClass):
    def __init__(self):
        super().__init__()
        print("Here's some additional logic that I want this subclass to have") …

我一直在学习d3,我对选择感到有些困惑.请考虑以下示例:

http://bl.ocks.org/mbostock/1021841

具体来说,让我们来看看这一行:

var node = svg.selectAll(".node")
    .data(nodes)
  .enter().append("circle")
    .attr("class", "node")
    .attr("cx", function(d) { return d.x; })
    .attr("cy", function(d) { return d.y; })
    .attr("r", 8)
    .style("fill", function(d, i) { return fill(i & 3); })
    .style("stroke", function(d, i) { return d3.rgb(fill(i & 3)).darker(2); })
    .call(force.drag)
    .on("mousedown", function() { d3.event.stopPropagation(); });

在文档中,它说:"选择是从当前文档中提取的一系列元素." 我解释这意味着svg.selectAll(.node)创建一个.node从当前文档中拉出的类元素数组,但据我所知,没有这样的元素!除非我感到困惑 - 而且我几乎可以肯定我是 - 文档中唯一给出了某个东西的地方"节点"是在选择已经发生之后(当我们写的时候.attr("class", "node")).

那么这里发生了什么？什么是svg.selectAll(".node")实际选择？

假设您正在创建一个包含四个子包的 python 包，并且其中generic_stuff.py两个子包恰好需要一个模块。具体来说，可能generic_stuff.py包含一些用于读取和写入数据库的自定义函数，子包 1 和 2 包含要单独部署的 Web 服务器的代码。

问题是：

使子包 1 和 2 可用的正确方法是什么？generic_stuff.py

我可以想到三种方法来做到这一点，但所有这些方法都感觉像是黑客：

1. 制作generic_stuff.py自己的包的一部分并安装它。这是不能令人满意的，因为它污染了整个系统的命名空间（更不用说子包3和4了）。
2. 在导入子包 1 和 2 之前，将父目录（包含generic_stuff.py四个子包）添加到系统路径。这是不令人满意的，因为它将父目录中的所有内容添加到子包 1 和 2 的命名空间中。这可以通过放入来generic_stuff.py避免其自己的目录并将其添加到系统路径中，但即使如此，如果子包 1 和 2 中的多个模块正在使用多个通用模块（想象一下尝试重构该项目），这种方法也无法很好地扩展。
3. 维护两份副本generic_stuff.py：一份位于子包 1 中，一份位于子包 2 中。我需要解释为什么这是一个坏主意吗？

我想我想到的是某种配置，也许会进入父目录__init__.py或其他东西，这将使模块 X 可用于子包 Y。是否存在类似的东西？

我有一堆代码:

statement1(args)
statement2(args)
statement3(args)
statement4(args)
statement5(args)

我想将语句分成块并在每个块之后写入日志.日志记录有点复杂:我想记录块执行后的每个块的运行时间和特定数据结构的状态.所以我创建了一个名为log_block处理所有这些细节的装饰器.现在我的代码看起来像这样:

@log_block()
def block1():
    statement1(args)
    statement2(args)

@log_block()
def block2()
    statement3(args)

@log_block()
def block3():
    statement4(args)
    statement5(args)

block1()
block2()
block3()

这很好用,但它有点笨重.令人讨厌的是我必须单独调用三个块函数,如果我想在块之间共享变量,那么我要么必须给出块函数参数和返回语句,要么使用全局变量,这两者都不是特别适合.我真正想要的是这样的语法:

@log_block()
    statement1(args)
    statement2(args)

@log_block()
    statement3(args)

@log_block()
    statement4(args)
    statement5(args)

所以我直接装饰语句而不是将它们包含在辅助块函数中.有没有办法实现这样的事情？

假设你有一个CSV有三列：item，username，和userid。使用Spark的Dataset API读取以下内容非常简单：

case class Flat(item: String, username: String, userid: String)
ds = sparkSession.read.csv("path/to/data").toDF("item", "username", "userid").as[Flat]

然后ds将为类型Dataset[Flat]。

但是，假设你希望你的数据有形式Dataset[Nested]，其中Nested由下式给出：

case class User(name: String, id: String)
case class Nested(item: String, user: User)

一种实现方法是将数据读入a Dataset[Flat]，然后应用a map将其转换为a Dataset[Nested]，但实际上，Flatcase类别通常不需要其他任何东西，并且它会使代码不必要地冗长。有什么方法可以跳过中间人并直接构造一个Dataset[Nested]？

假设我有一个类,我不希望每五秒钟调用一次以上的方法(例如,它会查询将过于频繁发送查询的应用程序列入黑名单的Web服务器).实现这个的最佳方法是什么？

我想出的解决方案是让方法在每次通过调用运行时在静态成员变量中存储时间戳System.currentTimeMillis(),并在执行时间敏感代码之前检查此时间戳.但是我很紧张,有些突发事件我没有预见到,所以我想看看是否有一种标准的方法来解决这个问题.

考虑以下简单的烧瓶应用:

from flask import Flask, request, session

application = Flask(__name__)
application.secret_key = "some_random_string"

@application.route("/enter_string")
def start_session():
    session["string"] = request.args["string"]

@application.route("/get_string")
def continue_session():
    if "string" not in session:
        return "Give me a string first!"

    return "You entered " + session["string"]

if __name__ == "__main__":
    application.debug = True
    application.run()

这是我的问题:

1. 一旦访问了"enter_string"端点并且用户已经分配了一个字符串session["string"],那么字符串存储在哪里？它是在服务器的内存还是用户的内存中？
2. 默认情况下,会话在浏览器退出时到期.是否有一种简单的方法让其他事件触发会话过期,例如关闭窗口但不一定是浏览器？
3. 默认情况下,会话是否会超时或是否一直保持到浏览器退出无论需要多长时间？

我想为我的项目创建一个记录器，它有一个跨所有子模块工作的自定义处理程序，并记录未捕获的异常。以下是我的概念验证；请注意，我使用文件处理程序作为最终的自定义处理程序的替代。

主要.py：

import logging
import sys
from module import divide

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

fh = logging.FileHandler("test.log")
fh.setLevel(logging.INFO)
logger.addHandler(fh)

def uncaught_error(exc_type, exc_value, exc_traceback):
    logger.error("Uncaught Exception", exc_info = (exc_type, exc_value, exc_traceback))
    sys.__excepthook__(exc_type, exc_value, exc_traceback)
    return

sys.excepthook = uncaught_error

if __name__ == '__main__':
    logger.info("first division problem")
    print(divide(5,2))
    logger.info("second division problem")
    print(divide(10,0))
    logger.info("end")

模块.py:

import logging

logger = logging.getLogger(__name__)

def divide(a, b):
    logger.info("dividing " + str(a) + " by " + str(b))
    return a/b

运行 main.py 后，这是 test.log 的输出：

first division problem
second …