Alexey Trukhanov's questions

鉴于

有许多函数可以执行一些计算。在某些情况下（为了调试或监控过程），定期打印计算结果是很方便的。像这样：

def some_calculations(b, a):
    print('Начали вычислять...')
    # чего-то вычисляем
    s = b + a
    print(f'Закончили вычислять. Результаты: {s}')

    print('Опять начали вычислять...')
    # чего-то вычисляем
    p = b * a
    print(f'Закончили вычислять. Результаты: {p}')
    
    # и тут таких или подобных итераций несколько
    
    return s, p

但在某些情况下，为了不浪费时间在长时间的输出操作上，我需要禁用所有这些打印。首先，我想定义某种可以传递给这些函数的布尔变量，并在每次打印附近描述根据这个布尔变量是否输出/不输出的条件。但后来我想一定有办法可以关闭所有的打印。我没有依靠自己的知识，而是去寻找并找到了它。

# выключили
sys.stdout = open(os.devnull, 'w')
# включили
sys.stdout = sys.__stdout__

接下来，我决定测量执行时间

1. 只是功能
2. 输出关闭
3. 打印被包装在取决于布尔标志变量的条件中的函数。

以下是带有标志的版本：

def some_calculations(b, a, flag = False):
    if flag:
        print('Начали вычислять...')
    # чего-то вычисляем
    s = b + a
    if flag:
        print(f'Закончили вычислять. Результаты: {s}')

    if flag:
        print('Опять начали вычислять...')
    # чего-то вычисляем
    p = b * a
    if flag:
        print(f'Закончили вычислять. Результаты: {p}')

    # и тут таких или подобных итераций несколько

    return s, p

测量结果表明，关闭输出的选项可使打印函数的执行时间缩短5 倍，而带有标志的选项可使执行时间缩短 65 倍。造成这种差异的原因很明显——开头指出的方法不会禁用打印，它仍然执行功能print，仅输出到devnull，也就是说，它只是抑制输出到控制台。并且标志绕过打印功能。

问题

是否有可能禁用某个函数（例如打印），而无需将每个函数调用包装在条件中，从而使禁用实际上绕过该函数，从而提供运行时好处，而不是简单地抑制输出？

笔记

在此任务中，您可以使用任何数据类型和模块，包括来自第三方开发人员的数据类型和模块（numpy, pandas, sorted containers等等）。主要标准是执行速度。

鉴于：

我们的逻辑的输入是连续提供相同结构的字典，其中包含某个事件的信息（参数）。在这些字典中，几个值是类型float（我们称之为键param_1, param_2...），其中一个值是该事件的时间戳（我们称之为键dt）。我们将这本词典分析成不同的列表。每个参数都在单独的列表中，时间序列也在单独的列表中。因此，在不同列表中的同一索引下存在有关同一事件的信息。我们知道时间戳是不递减的，这意味着下一个时间戳可以等于或大于前一个时间戳。相邻标记之间的时间增量是有条件随机的，可以是微秒，可以是几分钟，也可以有条件地不存在。

我们还给出了一个时间间隔（我们称之为duration）

任务

为了进一步处理，我们需要上述列表包含有关duration最后一个传入事件内的事件的信息。所有早于此时间的事件均被删除。

我的决定

我们接受吧pandas.Dataframe。事件到达——我们将字典分布在数据框的各列中。dt使用 mask按列进行过滤dt > время последнего события - duration。我们将数据框分成列表。

为了清楚起见，我编写了一个流程模拟器：

import time
import datetime
import random
import pandas as pd

random.seed(123)
duration = 30
df = pd.DataFrame(columns=['Param_1', 'Param_2', 'Param_3', 'dt'])

for _ in range(10000):
    # получили словарь
    input_dict = {'Param_1': random.random(),
                  'Param_2': random.random(),
                  'Param_3': random.random(),
                  'dt': datetime.datetime.now()}

    # добавили в датафрейм
    df.loc[0 if df.empty else df.index[-1]+1] = input_dict.values()

    # отфильтровали
    df = df.loc[df['dt'].gt(df.dt.iloc[-1] - datetime.timedelta(seconds=duration))]

    # разобрали по спискам
    param_1 = df.Param_1.to_list()
    param_2 = df.Param_2.to_list()
    param_3 = df.Param_3.to_list()
    dt = df.dt.to_list()

    # это рандомная задержка, чтобы эмулировать случайную разницу между событиями
    time.sleep(random.random() * 3)

问题

还有其他更快的选择吗？我将非常感激任何有关代码优化的评论。

如果我们有这样的表达式：

2 + 2

或者任何其他我们没有在任何地方分配的值，那么 PyCharm 会告诉你

该声明似乎没有效果

但解释器不会抱怨，代码将被执行。

问题是：计算结果将会去往何处？它会被存放在某个地方，然后被垃圾工人运走，还是根本不会被存放在任何地方？

这个问题由此而生。简而言之，我使用postgresql NOTIFY / LISTEN命令来组织两个进程之间的消息。作为python和postgersql之间的接口，我使用psycopg提供以下等待通知解决方案的模块：

import psycopg
conn = psycopg.connect("", autocommit=True)
conn.execute("LISTEN mychan")
gen = conn.notifies()
for notify in gen:
    print(notify)
    if notify.payload == "stop":
        gen.close()
print("there, I stopped")

在这里，我们创建一个生成器，当通知到达时它会“打开”，并等待它在循环中触发。循环阻塞流程，如果您不以gen.close()某种条件的形式提供退出，则循环将无限期等待。

我的任务略有不同。我不需要等待所有的消息，而只需要等待第一条消息，至于哪一条消息都无所谓。一旦第一条消息到达，阻塞等待过程就应该继续。该方法conn.notifies()（其中conn是一个包含与数据库的连接的psycopg对象）有一个参数，该参数在收到n 条stop_after=n消息后停止等待消息。根据我的任务，我指定并期待一条消息。stop_after=1

问题。如何以最节省内存/时间/语法的方式等待该生成器启动。我有两个解决方案。第一个解决方案，正如开发人员所建议的，是通过循环。

gen = conn.notifies(stop_after=1)
for _ in gen:
    pass

由于我们不关心收到什么消息，我使用了一个未命名的变量，并且在循环中不执行任何操作。

第二种选择是创建一个列表：

gen = conn.notifies(stop_after=1)
_ = list(gen)

# или даже вот так
_ = [1 for _ in gen]

#можно даже вот так, вообще никуда не присваивать
[1 for _ in gen]

如何提高预判能力？

给出了相当庞大的大容量数据结构。我的程序的逻辑需要使用这些表中的特定选择。采样需要相当长的时间，为了加快过程，我决定需要一个临时表，我将在其中进行一次选择，然后仅与该临时表进行通信，定期更新它。但在阅读 PostgreSQL 文档后，我发现了一个名为MATERIALIZED VIEW的工具，根据我的理解判断，它正好适合我的目的。

问题

使用单独的临时表与使用物化视图有什么实际区别（根据您的经验和知识） 。我阅读了文档，我很清楚第一段中描述的差异。我对其他方面感兴趣，例如，什么会工作得更快或者什么实际上更方便。

任务是将一个表完全复制到另一个表，而接收表中包含的数据必须完全替换为新表。也就是说，旧数据必须被删除并用新数据替换。与操作系统的类比告诉我，应该有一个查询简单地将一个表覆盖在另一个表之上（即，如果我们将一个文件复制到另一个文件，那么我们不需要关心目标文件中的内容，它只是将被覆盖）。然而，我没有找到这样的机械师。您需要删除接收器并再次创建它，复制源：

DROP TABLE dest;
CREATE TABLE dest AS TABLE source;

或者删除目标的内容并复制源的内容：

TRUNCATE TABLE dest;
INSERT INTO dest (fields) SELECT fields FROM source;

两个问题：

1. 有没有一种方法可以更新数据而无需先删除以前的数据。
2. 如果没有这样的方法，那么我建议的哪种方法更好，为什么？也许还有另外一个？

各位同事，我对异步编程完全不懂，想搞清楚，求帮助。我编写了一个项目，使用该库psycopg2作为 Postgresql 的接口。我决定向项目添加异步，将模块更新到psycopg版本 3，支持异步，并编写了一个简单的示例来说明我的问题。

import asyncio
import psycopg
from datetime import datetime

async def main():

    async def open_connection():
        async with await psycopg.AsyncConnection.connect(
                dbname=dbname,
                password=password,
                host='localhost') as aconn:
            async with aconn.cursor() as cur:
                await cur.execute("select count(*) from table_1 join table_2 using (id)")
                print("\nВыполнили запрос к БД")
                return await cur.fetchall()

    async def print_monitor():
        while True:
            print(f'\r{datetime.now()}', end='')
            await asyncio.sleep(0)

    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(open_connection())
        task2 = tg.create_task(print_monitor())

asyncio.run(main())

在示例中，有两个协程 - 第一个查询数据库，第二个不断打印时间（只是一个时间计数器）。当数据库返回对请求的响应时，我们打印“Completed...”。

问题 （如果需要分成两个单独的问题，请写在评论中）

1. 在同步代码中，该方法cursor.fetchall()返回基本响应（通常是元组列表）。在异步版本中，它返回一个协程。如何得到结果？
2. 从代码中可以看到，数据库查询完成后，时间输出并不会停止。模块中asyncio或其他地方是否有特殊方法来接收协程执行的确认，并在完成后立即中断计数器。也就是说，我认为条件while应该包含某个标志，当执行对数据库的请求的协程时会触发该标志。当然，我可以引入nonlocal一个变量main()并将其用作标志，但有些东西告诉我必须有某种异步工具。

UPT

我稍微改变一下第一个问题。在所有文档和示例中都写到，在异步代码中cursor.fetchall()它应该返回结果。但对我来说，它返回一个协程，这意味着我写错了。什么？

同事们，假设表中有一个字段包含一些数据序列。我需要获取一定数量的较大对象，但按升序排序。我想到的方式是先按降序排序，然后取出一定数量的对象，再按升序排序。像这样的东西：

SELECT * FROM 
(SELECT * FROM table_name
ORDER BY field_name DESC
LIMIT 100)
ORDER BY field_name

问题这是唯一最好的方法吗？是否有可能以某种方式立即获取序列的尾部而不是头部？

同事们，我也看到过类似的问题，但我认为自从他们被问到以来，这些年来突然发生了一些变化。

有一个包含许多字段的表。可能有一张表，也可能有很多张表，我们可以通过一些唯一的字段将它们组合起来。最初，任务是获取不包含NaN任何字段的记录，但事实上，问题更广泛 - 相反，NaN可以有任何值。

也就是说，如果我按一个字段搜索，那么显然：

SELECT * FROM table_name
WHERE field_name != 'NaN'::NUMERIC;

如果我搜索所有字段，那么我需要以一种或另一种方式明确指示字段和表的名称。但没有这样的事情：

SELECT * FROM table_name
WHERE table_name.* != 'NaN'::NUMERIC;

（我知道——这行不通）

也就是说，要么手动列出字段的名称，要么使用脚本获取名称并动态生成它们，或者一些非常华丽的（在我看来）动态查询。在所有这些情况下，我更容易pandas轻松愉快地将所有桌子拖进去并摆脱它NaN。但是如果您可以通过 SQL 轻松愉快地完成此操作呢？

谢谢你！

同事们，在分析这个问题的时候，我遇到了一些pandas.merge我完全无法理解逻辑的行为。我将不胜感激我的同事们的解释。即使是一些文档的链接也足够了。

给定。

有两个数据框由一列组成，我们试图通过这一列合并它们。

import pandas as pd

dic = {'col':['1','2','3','4','5','6','7','8','9']}
dic_1 = {'col':['9','8','7','6','5','4','3','2','1']}
df_1 = pd.DataFrame.from_dict(dic)
df_2 = pd.DataFrame.from_dict(dic_1)
df_total = pd.merge(df_1, df_2, on='col', how='inner')
print(df_1)
print(df_2)
print(df_total)

如果至少一个数据帧中的值不重复，则一切都很好且符合逻辑

  col
0   1
1   2
2   3
3   4
4   5
5   6
6   7
7   8
8   9

但问题是这样的。

如果相同的值在两个数据帧中重复，则生成的数据帧将包含该值 m*n 次，其中 m 和 n 分别是第一个和第二个数据帧中该值的重复次数：

import pandas as pd

dic = {'col':['1','2','3','3','3','6','7','8','9']}
dic_1 = {'col':['9','8','7','6','5','3','3','2','1']}
df_1 = pd.DataFrame.from_dict(dic)
df_2 = pd.DataFrame.from_dict(dic_1)
df_total = pd.merge(df_1, df_2, on='col', how='inner')
print(df_1)
print(df_2)
print(df_total)

结果如下：

   col
0    1
1    2
2    3
3    3
4    3
5    3
6    3
7    3
8    6
9    7
10   8
11   9

所有结果都是凭经验获得的，我只能对逻辑做出假设。

我将不胜感激的解释。

同事！给出了几个采用相同参数的函数。

def func1(a):
    print('Первая функция')
    return a

def func2(a):
    print('Вторая функция')
    return a

脚本从某个地方（例如，从存储配置的文件或从标准输入，这并不重要）接收有关运行哪些函数和不运行哪些函数的设置。

dic_flag = {'fun1': True, 'fun2': False}

您可以编写一堆条件语句来运行这些函数。

if dic_flag['fun1']:
    print(func1(a))

if dic_flag['fun2']:
    print(func2(a))

但我想自动化这个过程，例如，像这样：

def func1(a):
    print('Первая функция')
    return a

def func2(a):
    print('Вторая функция')
    return a

dic_flag = {'fun1': True, 'fun2': False}
dic_func = {'fun1': func1, 'fun2': func2}

a = 10

for flag in dic_flag:
    if dic_flag[flag]:
        print(dic_func[flag](a))

问题

你能为这项任务想出什么漂亮的东西？也许是 OOP 中的一些东西？或者有没有标准库？实际上有几十个功能。

选修的：

如果主要参数相同，但一些附加参数不同怎么办？

同事！给定一个具有列表或字典类型属性的类。任务：更改此列表/字典的元素之一，但尽管我们可以通过字符串变量或字符串常量使用属性名称。我知道使用内置对象的一种方法__dict__：

class MyClass:
  def __init__(self):
    self.attr: dict[str, int] = {'key1': 5,
                                 'key2': 7,
                                 'key3': 42}


my_class = MyClass()
attr_name = 'attr'
print(my_class.attr['key2'])

my_class.__dict__[attr_name]['key2'] = 12
print(my_class.attr['key2'])

问题：
这种方法在多大程度上可以被接受？还有其他更合适的方法吗？

各位同事，我在学习中dataclasses，遇到一个无法理解/实现的点。

一般信息。

如果不使用库本身的对象，就dataclasses不可能直接创建可变类型的类属性dataclasses。为什么这样做在文档中有描述，我也找到了；同事们正在详细讨论这个问题。但借助方法dataclasses还是可以做到的。

以下是不该做的事情的典型示例：

@dataclass
class Foo:
    bar: list = []

它将发生在这里ValueError

但这是你可以（应该！）：

@dataclass
class Foo:
    bar: list = field(default_factory=list)

在本例中，创建了 type 的属性List，但未分配默认值。文档没有描述如何在定义类时立即分配可变类型的默认值，但同事建议：

@dataclass
class Foo:
    bar: list = field(default_factory=lambda: [1, 2, 3])

问题。

有dataclasses一种方法.make_dataclass()可以动态定义数据类；它不需要提前定义。一般来说，我也很清楚它是如何工作的，但为可变类型的变量分配默认值是一个问题。我尝试了不同的方式，得到了不同类型的异常。

例如：

from dataclasses import make_dataclass, field
Config = make_dataclass('Config', [('x', field(default_factory=lambda: [1,2,3]))])
config = Config()
print(config.x)

在实例化创建时config = Config()会引发异常：

类型错误：Config.__init__() 缺少 1 个必需的位置参数：'x'

也就是说，我们创建了属性，但没有分配值。如果删除，会发生完全相同的事情lambda：创建类时没有错误，创建属性，但忽略默认值。

问题

如何使用具有可变类型属性的dataclass方法进行创建，该属性将立即被分配默认值。.make_dataclass()这原则上可能吗？

同事们，免责声明。有一个问题，从实现的角度来看，我想出了一个抽象且完全愚蠢的例子。这个例子只是为了说明我的问题，并不是一个好的解决方案的例子。

这是一个例子：

import random

random.seed(123)

def test_random(len):
    return [random.randint(0, 10) ** 2 for _ in range(len)]

print ([test_random(10)[i] for i in range(3)])

有一个函数可以生成并返回某个集合，可以通过索引/键来访问该集合的元素。列表、字典、元组 - 没关系，在示例中，为了便于说明，我使用了列表。给定相同的输入数据，该函数返回相同的结果。我们使用列表/字典包含，尝试根据此函数返回的集合生成一些其他集合。问题很明显 - 通过这种实现，我们每次迭代列表包含时都请求该函数，而不是一次获取对象并访问它。

问题。

是否可以在不多次调用函数并且不首先引入一个临时变量（我将在其中返回函数结果）的情况下实现这样的逻辑？

同事们，有一个表，比如说table，它有字段： id- 一个唯一的整数，value1, value2...valueN- 一些值。我使用来自 Python 的 API 请求从第三方服务接收新数据，其中包括id更新的值value。id在新数据中，它们可能包含在id表中的一组中，或者它们可能是新的。数以千计的此类更新记录可以同时到达。问题是如何最有效地替换表中存在的内容并添加不存在的内容。我看到以下选项：

第一个选项 使用 Python，将新的集合分为id表中已存在的和尚不存在的。然后分两个阶段：

阶段1。使用 Python 循环中动态生成的每条记录的更新来向数据库发送垃圾邮件。像这样的东西：

UPDATE table
SET value2=new_value_2, value5=new_value_5
WHERE id = id_from_python;

其中SET和 every中的新值id_from_python将在循环中被 python 替换。

第二阶段。为他们提供INSERT所有新的id数据。

INSERT INTO table (id, value1, value2, value3) 
VALUES (new_id, new_val, new_val, new_val), 
       (new_id, new_val, new_val, new_val);

第二个选择

我在这里读到的。简而言之，我们根据新数据创建一个临时表，并按照上面的链接进行操作，仅分两个阶段 -

阶段1。我们将工作表中的现有值更改为临时表中的值。

UPDATE table b
SET value1 = a.value1,
    value2 = a.value2,
    value3 = a.value3
FROM temp_table a
WHERE a.id = b.id;

第 2 阶段将临时表中的新表添加id到工作表中。

INSERT INTO table (id, value1, value2, value3)  
SELECT id, value1, value2, value3
  FROM temp_table
 WHERE temp_table.id not in table.id;

混合版本 使用Python，我们分为id新版本和现有版本。从现有的表中，我们创建一个临时表并按照第二个选项中的方式id执行操作，然后将许多新表添加到工作表中，就像第一个选项中一样。UPDATEid

问题。查询写得是否正确？哪个选项更好？有更好/更快/最佳的选择吗？

同事，我遇到了一个无法理解的方法行为，pandas.series.rolling请解释。为了研究，让我们采用以下人工代码：

import pandas as pd


def rolling_print(x):
    print(x)
    return float('nan')


ser = pd.Series([1, 2, 3, 5, 6, 7, 8, 9])
print(ser)
ser.rolling(5).apply(rolling_print)

解释一下：我们取pandas.series1int到9，把window 5的方法应用到这一系列rolling（window的值是多少无所谓，sint和s我都试过了datetime），然后apply用help调用函数显示每个窗口的数据。所以，在输出的时候，很明显数据已经由 变成int了float。即1.0, 2.0...，等。为什么？float('nan')（用任何类型常量替换返回的数据并int没有改变结果）

各位大侠，有一个关于优化的问题。给定一个元组列表，例如：

a = [(1,5,6), (7,8,9), (10,12,4)]

我马上预定，现实生活中元组有几百万个，每个元组有七个元素，这个例子只是一个例子，为了便于理解。

有必要将这个元组列表转换为列表字典，以便元组的所有零元素成为一个列表，第一个元素 - 下一个列表，第三个 - 下一个......键将被手动设置。像这样：

a_d = {'key1':[1,7,10], 'key2':[5,8,12], 'key3':[6,9,4]}

我知道如何解决这个问题Pandas。是这样的：

import pandas as pd
a = [(1,5,6), (7,8,9), (10,12,4)]
a_d = {key: list(val) for key, val in dict(pd.DataFrame(a, columns=['key1', 'key2', 'key3'])).items()}

有两个问题：

1. 有没有一种方法可以使用基本的 Python 和标准库来解决这个问题，而无需遍历所有元素。也许我缺少一些标准方法/功能。
2. 这种方法会比上面描述的方法更快吗Pandas？

谢谢。