Когда требуется быстро расшарить файлы в локальную сеть со своего компа можно использовать дефолтный python-сервер. Все решается одной командой.

python3 -m http.server

Но это бывает неудобным если нужно скачать папку или залить файлы. В этом случае более удобным будет быстрый FTP сервер.

Я себе сделал шорткат для поднятия простого FTP сервера без авторизации на базе библиотеки pyftpdlib.

Варианты запуска:

# на рандомном порту read only
python3 -m pyftpdlib 

# на указанном порту
python3 -m pyftpdlib -p 22222

# с доступом на запись
python3 -m pyftpdlib -w

# с авторизацией
python3 -m pyftpdlib -w --user=name --password=123

# полный список аргументолв
python3 -m pyftpdlib -h

Мой алиас для расшаривания в текущей директории

alias ftp="python3 -m pyftpdlib -w -p 22222"

Теперь можно подключть FTP соединение как удалённую директорию стандартными средствами OS. В Windows это Add Network Location, в Linux - зависит от дистрибутива. Ищите в разделе Network вашего файлового браузера.
Также можно использовать сторонние клиенты, например FileZilla.

А здесь подробней про http.server

#libs #tricks

Регулярно приходится писать и ревьюить код, где используется PySide2-6.
Заметил, что в подавляющем большинстве случаев настройка создаваемых базовых виджетов происходит через методы. Думаю, всем знаком такой способ.

Простой пример с кнопкой:

button = QPushButton("Click Me")
button.setMinimumWidth(300)
button.setFlat(True)
button.setStyleSheet("font-size: 20pt")
button.setToolTip("Super Button")
button.clicked.connect(lambda: print("Button clicked"))

Но есть и альтернативный способ - настройка через свойства. Это просто ключевые аргументы конструктора класса. Хоть они и не указаны в документации как аргументы, но они есть)

Этот код делает тоже самое но с помощью Property

button = QPushButton(
    "Click Me",
    minimumWidth=300,
    flat=True,
    styleSheet="font-size: 20pt",
    toolTip="Super Button",
    clicked=lambda: print("Button clicked"),
)

Где это может быть полезно

▫️ Это выглядит более аккуратно и коротко, уже повод использовать

▫️ Может использоваться в заполнении лейаута, когда нам не нужно никакое другое взаимодействие с виджетом и поэтому сохранять его в переменную не требуется. Например, лейбл или кнопка.

widget = QWidget(minimumWidth=400)
layout = QHBoxLayout(widget)
layout.addWidget(QLabel("Button >", alignment=Qt.AlignRight))
layout.addWidget(QPushButton("Click Me", clicked=lambda: print("Button clicked")))
widget.show()

Либо так

widget = QWidget(minimumWidth=400)
layout = QHBoxLayout(widget)
for wd in (
        QLabel("Button >", alignment=Qt.AlignRight),
        QPushButton("Click Me", clicked=lambda: ...)
    ):
    layout.addWidget(wd)
widget.show()

▫️ Можно хранить настройки в каком-то конфиге или генерировать на лету, после чего передавать как kwargs.

kwargs = {"text": "Hello " * 30, "wordWrap": True}
my_label = QLabel(**kwargs)

Как получить полный список доступных свойств?

Эта функция распечатает в терминал все свойства виджета и их текущие значения

def print_widget_properties(widget):
    meta_object = widget.metaObject()
    for i in range(meta_object.propertyCount()):
        property_ = meta_object.property(i)
        property_name = property_.name()
        property_value = property_.read(widget)
        print(f"{property_name}: {property_value}")

#tricks #qt

Установить свойства виджета в PySide можно не только через соответствующие методы и конструктор класса. Можно их изменять с помощью метода setProperty по имени.

btn = QPushButton("Click Me")
btn.setProperty("flat", True)

Это аналогично вызову

btn.setFlat(True)

Если указать несуществующее свойство, то оно просто создается

btn.setProperty("btnType", "super")

Получить его значение можно методом .property(name)

btn_type = btn.property("btnType")

Когда это может быть полезно?

▫️Можно просто хранить какие то данные в виджете и потом их доставать обратно

widget = QWidget()
widget.setProperty('my_data', 123)
print(widget.property('my_data'))

▫️ Назначая эти свойства разным виджетам можно потом отличить виджеты во время итераци по ним. Например, найти все кнопки со свойством my_data="superbtn".

Но ведь вместо кастомного свойства можно использовать objectName, будет тот же результат.

Да, но y ObjectName есть ограничение - только строки.

▫️ Если нам потребуется не просто поиск а, например, сортировка по числу, то свойства позволяют нам это сделать. Поддерживается любой тип данных

widget.setProperty('my_data', {'Key': 'value'})
widget.setProperty('order', 1)
all_widgets.sort(key=w: w.property('order'))

Но ведь Python позволяет всё вышеперечисленное сделать простым созданием атрибута у объекта

widget.order = 1
widget.my_data = 123

Да, но я думаю что не надо объяснять почему не стоит так делать. К тому же, если у виджета нет свойства то метод .property(name) вернет None, а отсутствующий атрибут выбросит исключение.

▫️ Действительно полезное применение кастомным свойствам - контроль стилей. Здесь атрибутами не обойтись, нужны именно свойства.
Дело в том, что в селекторах стилей можно указывать конкретные свойства виджетов на которые следует назначать стиль.

Просто запустите этот код

from PySide2.QtWidgets import *

if __name__ == "__main__":
    app = QApplication([])

    widget = QWidget(minimumWidth=300)
    layout = QVBoxLayout(widget)
    btn1 = QPushButton("Action 1")
    btn2 = QPushButton("Action 2")
    btn3 = QPushButton("Action 3", flat=True)
    layout.addWidget(btn1)
    layout.addWidget(btn2)
    layout.addWidget(btn3)
    # добавим кастомное свойство одной кнопке
    btn1.setProperty("btnType", "super")
    # добавляем стили
    widget.setStyleSheet(
        """
        QPushButton[btnType="super"] {
            background-color: yellow;
            color: red;
        }
        QPushButton[flat="true"] {
            color: yellow;
        }        
    """
    )
    widget.show()
    app.exec_()

С помощью селектора мы избирательно назначили стили на конкретные кнопки.

Как получить список всех кастомный свойств?

Функция получения списка кастомных свойств отличается от получения дефолтных.

def print_widget_dyn_properties(widget):
    for prop_name in widget.dynamicPropertyNames():
        property_name = prop_name.data().decode()
        property_value = widget.property(property_name)
        print(f"{property_name}: {property_value}")

#tricks #qt

Как добавить директорию в игнор git репозитория.

1. Те, кто работает JetBrains-продуктах уже на автомате добавляют в .gitignore строчку: .idea/. Это самый простой способ.

2. Чтобы не добавлять в каждом проекте можно добавить в глобальный игнор файл. По умолчанию он лежит здесь:

~/.config/git/ignore

Либо указать другой путь через конфиг

git config --global core.excludesfile ~/.gitignore

Кстати, библиотека venv в Python 3.13 по умолчанию в корень вирутального окружения добавляет файл .gitignore с одним символом *, что означает исключение всего в текущей директории. Таким образом папка с venv автоматически исключается из репозитория. Удобно.

#tricks

Недавно возникла такая задача: требовалось из Python скрипта запустить дочерний процесс, тоже Python скрипт, и получить от него некоторые данные. В моём случае это был некий словарь который мог быть сериализован в JSON формат, но это не так важно.

Какие есть варианты это сделать?

1️⃣ Передать дочернему процессу путь к файлу куда и будет записан результат.
После завершение дочернего процесса просто читаем данные из файла.

✅ легко и понятно, все так умеют делать
✅ можно перемещаться по файлу через seek
✅ можно прочитать когда-нибудь потом
❌ обращение к файловой системе, бывает относительно не быстро
❌ какое-то время файл будет доступен любому процессу, небезопасно
❌ только полная запись данных перед чтением (на самом деле есть вариант чтения во время записи, но это не то что мы хотим делать😖)


2️⃣ TCP/UDP сокет

✅ универсально, даже для неродственных процессов
✅ нет обращения в файловой системе (Unix-сокеты это почти файлы но всё равно не совсем)
✅ можно стримить данные
❌ нужна какая-то система авторизация чтобы обезопасить доступ
❌ оверхед для простой передачи данных, особенно если процесс дочерний. Требуется поднятие сервера и организция клиента со всеми вытекающими зависимостями и конструкциями


3️⃣ Парсить аутпут дочернего процесса.

✅ быстро, так как пайпы работают через оперативную память
✅ нет обращения к файловой системе и всех действий с этим связанных
✅ пайп привязан к файловым дескрипторам конкретных процессов, и доступ к нему могут получить только те процессы, которые унаследовали этот дескриптор (или получили другим способом)
✅ передача данных в режиме стрима
❌ неудобно если дочерний процесс пишет логи в stdout, нужна какая-то логика выделения только нужного или как-то отключать логи в надежде что никто другой туда ничего не напишет.
❌ нельзя перемещаться через seek

Если у вас взаимодействие с дочерним процессом, то есть самый простой вариант - кастомный пайп!✨

Это как stdout или stderr, но только еще один канал в котором не будет никаких логов и сообщений об ошибках.
Для простоты примера сделаем один пайп. Дочерний процесс должен что-то прислать в родительский процесс.

👮‍♂️РОДИТЕЛЬСКИЙ ПРОЦЕСС

1. Создаем новый пайп

import os. subprocess

read_fd, write_fd = os.pipe()
# важный момент! добавляем возможность наследовать дескриптор дочерним процессом. Обязательно после Python 3.4+ (PEP 446)
os.set_inheritable(write_fd, True)

2. Запускаем дочерний процесс передавая ему номер файла

process = subprocess.Popen(
    [sys.executable, child_script, str(write_fd)],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,
    close_fds=False # важный момент! это нужно, чтобы дочерний процесс сохранил все открытые дескрипторы, а не только стандартные потоки
)
os.close(write_fd) # закрываем дескриптор чтобы у родителя не висел открытый конец записи, иначе в читающем конце не наступит EOF

3. Читаем данные

with os.fdopen(read_fd, 'r') as data_pipe:
    data = data_pipe.read()
print('RECEIVED:', data)

Чтение прекратится когда файл закроется, за это отвечает контекстный менеджер with в дочернем процессе.

Стандартные пайпы тоже можно прочитать

stdout_log, stderr_log = process.communicate()
print(stdout_log)
print(stderr_log)

👶 Переходим к коду дочернего процесса.

1. Получаем номер дескриптора

write_pipe_fd = int(sys.argv[-1])

Пишем в него данные

with os.fdopen(write_pipe_fd, 'w') as data_pipe:
    data_pipe.write('Hello!')
    data_pipe.flush()

Вот и всё, мы сделали коммуникацию между двумя процессами через кастомный пайп ⭐️
Быстро, легко, безопасно!

С помощью двух пайпов можно ораганизовать передачу сообщений между процессами в обе стороны.

Пример с JSON можно глянуть здесь↗️

#tricks

Быстрый встроенный профайлинг на Linux с помощью time

time python -c 'for i in range(10**7): i**2'

Покажет время выполнения процесса

real    0m2,470s
user    0m2,405s
sys     0m0,074s

real - Общее время, прошедшее с момента запуска до завершения программы. Включая время ожидания I\O или переключения контекста.
user - Количество времени, которое CPU потратил на выполнение кода самой программы в пользовательском режиме.
sys - Количество времени, которое CPU потратил на выполнение системных вызовов (операций ядра, таких как чтение/запись файлов, управление памятью) от имени программы.

Но это встроенная команда из моей оболочки. Есть такая же GNU-утилита и она может показывать больше информации. Но нужно вызывать по абсолютному пути, так как builtin команда имеет бОльший приоритет.

/usr/bin/time -v python -c 'for i in range(10**7): i**2'

Command being timed: "python -c for i in range(10**7): i**2"
User time (seconds): 2.38
System time (seconds): 0.07
Percent of CPU this job got: 100%
...

Кроме времени исполнения будет также показано много другой полезной информации
- эффективность использования CPU (в %)
- максимальный объем занятой памяти
- обращения к файлам
- код выхода

И другие сведения.

#tricks

В работе с медиа файлами часто требуется определить не просто расширение, а его, скажем так, "категорию". Тоесть определить это видео, аудио или картинка. Примерно в 10 случаях из 10 в ревью я вижу обычный хардкодинг с большим мапингом и соответствующим поиском по нему.

file_type_by_ext = {
    'video': ['.mp4', '.mov', '.mkv', ...],
    'audio': ['.mp3', '.wav', '.ogg', ...],
    'image': ['.jpg', '.png', '.exr', ...]
}

Для таких случаев есть простой способ - стандартная библиотека mimetypes.

import mimetypes
mimetypes.guess_type("example.txt")
# ('text/plain', None)

Причём ей не нужен файл, достаточно просто имени строкой.

Первый элемент кортежа это MIME-тип (Multipurpose Internet Mail Extensions Type) - стандартный способ идентификации формата файла.

Формат: type/subtype

type - общая категория данных (text, video, image)
subtype - конкретный формат внутри категории

mimetypes.guess_type("photo.jpg")
# ('image/jpeg', None)
mimetypes.guess_type("render.mp4")
# ('video/mp4', None)

Второй элемент это тип кодировки содержимого, обычно для контейнеров типа gz и аналогичных.

mimetypes.guess_type("file.tar.gz")
# ('application/x-tar', 'gzip')
mimetypes.guess_type("backup.tar.bz2")
# ('application/x-tar', 'bzip2')

Итого, узнать категорию файла одной строкой:

mimetypes.guess_type('myfile.mov')[0].split('/')[0]
# video

Конечно при условии, что тип будет распознан, иначе будет None а не строка. Но об этом в следующий раз.

#libs #tricks

import mimetypes
mimetypes.guess_type("example.fbx")
# (None, None)

Формат не распознан, так как не зарегистрирован в системе.
Регистрация происходит с помощью функции mimetypes.init(). Эта функция автоматически вызывается при первом обращении.
Для каждой OS работает по-разному. В Windows читает реестр, в Linux достает всё из файла /etc/mime.types, в MacOS читает из системной БД.

На linux можно попробовать распознать тип через вызов

file --mime-type -b <filename>

эта команда попробует прочитать метадату самого файла, то есть должен быть доступ к файлу. Но это не гарантия успеха.

Можно попробовать использовать нестрогое соответствие IANA с помощью флага strict=False. Тогда будут учтены старые и нестандартные типы. Обычно они с префиксом x-

Новые типы можно добавлять самостоятельно.

mimetypes.add_type('application/x-fbx', '.fbx') # с точкой
mimetypes.guess_type("example.fbx")
# ('application/x-fbx', None)

Либо вызвать init() еще раз передав список текстовых файлов с нужными вам типами (без точки)

# my-mime-types.txt
application/x-fbx    fbx
application/x-ogo    ogo
application/x-aga    aga

mimetypes.init(['my-mime-types.txt'])
mimetypes.guess_type("example.ogo")
# ('application/x-ogo', None)

Есть и обратная операция - получить расширение файла из mime-типа

mimetypes.guess_extension('image/jpeg')
# .jpg

Или все подходящие расширения

mimetypes.guess_all_extensions('image/jpeg')
# ['.jpg', '.jpe', '.jpeg', '.jfif']

Советую почитать полную документацию
Также обратите внимание на библиотеку content-types для работы с mime-типами, где больше возможностей.

#libs #tricks

Все знают синтаксический сахар с операторами +=, -= и тд

x += 1

Где под капотом он превращается в

x = x + 1

Останется ли переменная х той же переменной после +=?
Конечно нет, это же неизменяемый тип

x = 1
print(id(x))
# 135373664533280
x += 1
print(id(x))
# 135373664533312

Теперь провернём тоже самое со списком

ls = [1, 2]
print(id(ls))
# 135373622585344
ls = ls + [3]
print(id(ls))
# 135373619036608

Ожидаемо работает так же, ведь мы создали новую переменную.
А теперь попробуем иначе:

ls = [1, 2]
print(id(ls))
# 135373622585344
ls += [3]
print(id(ls))
# 135373622585344
print(ls)
# [1, 2, 3]

И, внезапно, это работает не так как с int, со списками оператор += работает как extend()!
То же самое будет с *=, объект останется тем же.

ls = [1, 2]
print(id(ls))
# 135373622585344
ls *= 2
print(id(ls))
# 135373622585344
print(ls)
# [1, 2, 1, 2]

Следует помнить о такой важной разнице!
(Особенно на собесах 😉)

#tricks

Не запуская код определите, что покажет терминал если выполнить следующее:

_A__b = 'c'
class A:
    def get(self):
        return __b
print(A().get())

Ответ: Несмотря на то, что ваш IDE покажет ошибку, ошибки не будет. Распечатается "c"

Объяснение:

1. Mangling
За это отвечает механизм mangling - искажение имени. Так работают приватные атрибуты классов.
При создании атрибута по правилу: минимум 2 "_" в начале и максимум 1 "_" в конце" имя автоматически становится вида _{classname}{attr}
В нашем случае атрибутов класса не создается, но это не отменяет Mangling при обращении к другим объектам внутри класса.

2. Обращение к атрибуту
Когда внутри класса происходит обращение к любому объекту с именем по указанному выше правилу, его имя на уровне байт кода также преобразуется.

3. Поиск
Далее происходит поиск такой переменной по неймспейсам в порядке LEGB - Local, Enclosing, Global, Built-in.
И не трудно догадаться что мы находим нужный атрибут в Global, В итоге получаем результат!

Проверить можно так:

import dis
dis.dis(A.get)
# 4  RESUME       0
#
# 5  LOAD_GLOBAL  0 (_A__b)
#    RETURN_VALUE

Либо удалите переменную _A__b и запустите еще раз, поулчите ошибку:

NameError: name '_A__b' is not defined

Как думаете, это норма или баг?

#tricks

Потоковая обработка часто встречается при работе с большими файлами или когда данные приходят частями. В Python есть множество инструментов для работы с такими данными. Самый известный - итератор файла по строкам. В веб-приложениях это стандарт для передачи файлов. Далее приведу несколько примеров.

Чтение файлов

with open('huge-file.txt') as file:
    for line in file:
        process_line(line)

Это позволяет нам читать текстовый файл по строкам не загружая всё в память.
Конечно, если позволяет формат данных. С JSON такое не сработает (ijson может в этом помочь).

Запись файла чанками

with open('file-to-save.txt',
    'w') as file:
  for line in iter_data():
      file.write(line)

Частные случаи есть в разных библиотеках. Например DictWriter и DictReader из модуля csv позволяет работать с конкретным форматом данных а не просто текст.

import csv

with open('data.csv', 'r') as file:
  reader = csv.DictReader(file)
  for row in reader:
    print(row)

with open('data.csv', 'a', 
      newline='') as f:
  writer = csv.DictWriter(f, 
    fieldnames=['col1', 'col2']
    )
  for row in iter_objects():
    writer.writerow(row)

Отдельно интересен ZipFile, позволяющий "открыть" файл сразу внутри архива и записывать его частями

import zipfile as zf

with zf.ZipFile(
    'archive.zip', 
    'w', 
    compression=zf.ZIP_DEFLATED) as zf:
  with zf.open(
      'large_data.bin', 
      mode='w') as in_file:
    with open(
        'large_data.bin', 
        'rb') as source:
      for chunk in iter(
          lambda: source.read(1024), 
          b''):
        in_file.write(chunk)

Создание хеша для большого файла

import hashlib

sha256 = hashlib.sha256()
with open(
    'large-file.bin', 
    'rb') as f:
  for block in iter(
      lambda: f.read(1024), b''
    ):
    sha256.update(block)
hash_sum = sha256.hexdigest()

Сжатие данных в файл отдельными чанками

import gzip

with gzip.open('data.gz', 'wb') as f:
    for bin_chunk in iter_bin_data():
        f.write(bin_chunk)

Чтение с записью в файл

with gzip.open('data.gz', 'rb') as f_in:
  with open(
      'extracted_data.txt', 
      'wb') as f_out:
    for chunk in iter(
        lambda: f_in.read(1024), 
        b''):
      f_out.write(chunk)

Подсчет объектов из стрима. Добавление обновляет счетчики.

from collections import Counter

c = Counter()
for data in iter_objects():
    c.update(data)

Это не все доступные примеры, их еще много. Каждый из них позволяет обрабатывать данные из потока не ожидая весь набор и не загружая их в оперативку.
Это очень полезная техника, которую я призываю использовать по назначению!

#tricks #libs

reload_flag=""
if [[ -n "${DEBUG}" ]]; then
  reload_flag="--reload"
fi

if [[ -n "${WORKER_COUNT}" ]]; then
  workers=${WORKER_COUNT}
else
  workers=2
fi

gunicorn --workers ${workers} \
  --bind 0.0.0.0:8000 \
  ${reload_flag} main.wsgi

Писали такие конструкции чтобы проверить наличие флага и сформировать команду правильно?
На самом деле можно сделать тоже самое проще. Для этого используются операторы условной подстановки, доступные в оболочках семейства POSIX.

:- для установки значений по умолчанию

${WORKER_COUNT:-2}

Если переменная не объявлена, то будет дефолтное значение 2.

:+ подставляет указанный текст, если переменная не пуста

${DEBUG:+--reload}

Если что-то есть в переменной то распечатается текст после символа +, в противном случае - ничего. Удобно для опциональных флагов, как в нашем примере.

Итого наш скрипт может выглядеть так:

gunicorn --workers ${WORKER_COUNT:-2} \
  --bind 0.0.0.0:8000 \
  ${DEBUG:+--reload} main.wsgi

Есть еще два оператора.

:= не только подставить дефолтное значение, но и присвоить его переменной, если она пуста

# никаких переменных еще нет
VAL1=${VAL2:=hello}
# теперь доступны обе
echo $VAL1 $VAL2
# hello hello

:? остановить выполнение с ошибкой, если переменной нет.

echo ${MISS:?is required}
bash: MISS: is required

Код выхода будет 1.

#tricks #linux

Вы до сих пор используете в проекте "магические" строки?😖

@dataclass
class Task:
    status: str
    ...

def create_pending_task(data: dict) -> Task:
    task = Task(**data)
    task.status = "pending"  # < магическая строка
    return task

Где тут проблема?

🔸 Если "pending" изменится на "wait", вам придется искать это слово по всему проекту
🔸 Напишете panding вместо pending и баг вылезет только в рантайме в непредсказуемом месте
🔸 Вам очень повезет, если в проекте нет такой же строки но с другим смыслом

Как делать правильно?

Используем модуль enum

from enum import StrEnum

class TaskStatus(StrEnum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"

@dataclass
class Task:
    status: TaskStatus
    ...

def create_pending_task(data: dict) -> Task:
    task = Task(**data)
    task.status = TaskStatus.PENDING
    return task

Почему это лучше:
▫️Теперь это не строка а объект
▫️ IDE сможет подсказать какие статусы существуют, вам не нужно лезть в документацию или базу
▫️ Единый источник истины. Изменяем в одном месте вместо поиска на всему проекту
▫️ Типизация - наше всё, mypy умеет с этим работать
▫️ Читаемость кода повышается. Ведь читаем мы его чаще чем пишем
▫️ Автоматическая валидация допустимых значений в моделях Pydantic

#tricks

Почему в прошлом посте я использовал StrEnum а не Enum?
Всё просто, дефолтный Enum не поддерживает нативное сравнение с нужным нам типом.

from enum import Enum

class DefaultEnum(Enum):
    KEY = "value"

"value" == DefaultEnum.KEY # False
"value" == DefaultEnum.KEY.value # True

Как видите, приходится вызывать .value, что неудобно в некоторых случаях и более многословно. StrEnum это исправляет:

from enum import StrEnum

class StringEnum(StrEnum):
    KEY = "value"

"value" == StringEnum.KEY # True 

Для примера из прошлого поста это выглядело бы так:

if task.status == TaskStatus.PENDING:
    ...

Точно так же работает и IntEnum.

StrEnum появился в версии 3.11, для более ранних использовали комбинацию MyEnum(str, Enum), что не тоже самое.
StrEnum правильно создает значения с функцией auto(). Сочетание str+Enum создает числа, но в виде строк. Приходится явно писать строки. Сделал пару примеров для сравнения↗️

Когда не стоит использовать StrEnum:
- когда нужно явное отличие значений энума от строки
- когда в проекте уже используется обычный Enum

#tricks

Еще одно применение пайпов - в контексте с Enum.
Но для этого нужен специальный Enum основанный на типе Flag.
В связке с auto он генерирует битовые маски, которые впоследствии можно использовать с оператором |

from enum import Flag, auto

class Perm(Flag):
    READ = auto()    # 1 (0001)
    WRITE = auto()   # 2 (0010)
    EXECUTE = auto() # 4 (0100)
    DELETE = auto()  # 8 (1000)

Теперь мы можем комбинировать их через пайп

admin_perms = Perm.READ | Perm.WRITE | Perm.EXECUTE
user_perms = Perm.READ | Perm.EXECUTE
print(admin_perms)
# <Perm.READ|WRITE|EXECUTE: 7>

Можно делать проверки через in (возвращает bool)

if Perm.READ in admin_perms:
    print("Success!")

Либо через & (возвращает совпадение либо 0)

print(Perm.READ & admin_perms)
# <Perm.READ: 1>
print(Perm.WRITE & user_perms)
# <Perm: 0>

Оператор ~ инвертирует все флаги

print(~admin_perms)
#<Perm.DELETE: 8>

Можно заранее создать комбинацию.

class Perm(Flag):
    READ = auto()    # 1 (0001)
    WRITE = auto()   # 2 (0010)
    EXECUTE = auto() # 4 (0100)
    DELETE = auto()  # 8 (1000)
    RW = READ | WRITE

mode = Perm.READ
print(mode & Perm.RW)
# <Perm.READ: 1>  (True)
print(mode & Perm.EXECUTE)
# <Perm: 0>  (False)

Flag более изолирован. Он не равен числу напрямую, что защищает от случайных ошибок в логике.

#tricks

Если запустить REPL с модулем asyncio, то вы входите в особый асинхронный REPL.

user@host:~$ python -m asyncio
asyncio REPL 3.12.7 ...
Use "await" directly instead of "asyncio.run()".
>>> import asyncio
>>> 

В этом режиме
- создаётся и настраивается event loop
- уже импортирован asyncio
- работает await на верхнем уровне

То есть такая команда сработает без ошибок!

await asyncio.sleep(3)

Удобно для тестирования асинхронных функций без создания ивентлупов и остальной обвязки.

Работает в: 3.8+

#tricks #async

Теперь аналогичная история с тредами. Для тредов используется объект threading.local.
Он позволяет создать локальный динамический атрибут (да, вот так костыльно) для треда.

Вот базовый пример:

import threading
import time
import random

# глобальная переменная
thread_data = threading.local()

def execute():
    # поулчаем локальное значение для текущего треда
    current_user_id = getattr(thread_data, "user_id", -1)
    print(f"Log {threading.current_thread().name}: {current_user_id}")

def thread_task(user_id):
    # устанавливаем значение для текущего треда
    time.sleep(random.random())
    thread_data.user_id = user_id
    print(f"Create {threading.current_thread().name} == {user_id}")
    execute()

threads = [
    threading.Thread(
        target=thread_task, 
        args=(i,), 
        name=f"Thread-{i}")
    for i in range(10)
    ]
for t in threads:
    t.start()
for t in threads:
    t.join()

Вывод должен быть аналогичным, с соотетстивем номера треда и id юзера.

Есть еще один пример здесь

#tricks

Мы рассмотрели два способа управления конеткстом переменных. Если вам показалось, что это выглядит излишне и можно было бы оставить один, то вам не показалось.
Способ с threading.local придуман для разделения переменных между потоками. CоntextVar был добавлен как новый метод для асинхронного кода, но оказался настолько универсальным, что его можно использовать и с потоками.
После появления ContextVar в PEP567 его рекомендовано использовать вместо threading.local.
И даже был сделан бекпорт для версия ниже 3.7.1.

Теперь, если совместить ContextVar и Proxy-класс из прошлого примера то получим такой класс↗️.

Но у этого класса есть две проблемы:

1️⃣ Нигде не вызывается reset для сброса переменной, что может приводить проблемам

- утечка памяти
- "грязный" конеткст при переиспользовании потоков
- невозможность вернуться к дефолту

Решим это с помощью конектстного менеджера:

@contextlib.contextmanager
def configure_context(self, *args, **kwargs):
    """Синхронный контекстный менеджер (для `with`)"""
    tok_cfg = self._cv_config.set((args, kwargs))
    tok_obj = self._cv_object.set(None)
    try:
        yield self
    finally:
        self._cv_object.reset(tok_obj)
        self._cv_config.reset(tok_cfg)

@contextlib.asynccontextmanager
async def aconfigure_context(self, *args, **kwargs):
    """Асинхронный контекстный менеджер (для `async with`)"""
    tok_cfg = self._cv_config.set((args, kwargs))
    tok_obj = self._cv_object.set(None)
    try:
        yield self
    finally:
        self._cv_object.reset(tok_obj)
        self._cv_config.reset(tok_cfg)

Пример использования:

with proxy.configure_context(val1, val2):
    proxy.do_something()

Теперь прокси готов, но...

2️⃣ В асинхронном коде, для которого и придуманы ContextVar, созданием корутин занимается Event Loop, именно он отвечает за наследование контекста дочерними корутинами. В случае с потоками ничего такого нет, мы сами себе "эвентлуп", поэтому приходится прописывать копирование конеткста самстоятельно.

Пример проблемы с отсутствием наследованием конеткста в потоках↗️

Для решения есть функция копирования текущего контекста и метод запуска функции с новым конектстом:

сontextvars.copy_context().run(func, *args, **kwargs)

Здесь сложно придумать универсальное автоматическое копирование контекста, самая простая функция будет выглядеть так:

def run_in_thread_with_context(
        func: Callable, *args, **kwargs
        ) -> threading.Thread:
    ctx = contextvars.copy_context()
    t = threading.Thread(
        target=lambda: ctx.run(func, *args, **kwargs)
        )
    t.start()
    return t

И если вернуться к нашему синхронному ApiClient, то придётся следить за конектстом самостоятельно. И если где-то в коде библиотеки уже есть вызов тредов, то это работать не будет, придется переписывать.

threading.local тоже не наследует конеткст.

Полный пример Proxy с CоntextVar↗️

Пример использования:

client = ContextVarProxy(ApiClient)

def worker_in_thread(token):
    with client.configure_context(token=token):
        use_client(...)

Еще вариант, это кастомные ThreadExecutor и Thread с поддержкой автокопирования контекста. Забираем здесь↗️

И нет, это не пример как надо делать в проде) Это просто эксперемент для понимания процесса.

#tricks

Как-то давно писал трансфер файлов по сети.
В этом проекте требовалось создавать файл, который сразу существует на диске, имеет нужный размер но еще не содержит данных.
Вот примеры как создать такой файл:

length = 1024 * 1024 * 1024 * 100
with open(file_path, "wb") as out:
    out.seek(length-1)
    out.write(b"\0")

with open(file_path, "wb") as out:
    out.truncate(1024 * 1024 * 1024 * 120)

truncate -s 100M test

Файл создается моментально и получается полностью состоящий из нулей. Более того, он не занимает место над диске!
Это называется sparse files - разреженные файлы. На таких файловых системах как ext4, XFS, Btrfs, ZFS файл автоматически становится разреженным если процесс пишет за пределы конца файла. В структуре файла создаются "дырки" которые автоматически при чтении вернут нули.

Если запустить тоже самое на Windows, то результат будет другой. Файл будет создаваться долго и реально займет место на диске.

NTFS умеет создавать разреженные файлы, но это надо активировать явно:

import os
import msvcrt
import ctypes

file_path = r"C:\file"
length = 1024 * 1024 * 1024 * 100  # 100 GB

with open(file_path, "wb") as f:
    handle = msvcrt.get_osfhandle(f.fileno())
    FSCTL_SET_SPARSE = 0x900C4
    bytes_returned = ctypes.c_ulong()
    ctypes.windll.kernel32.DeviceIoControl(
        handle, FSCTL_SET_SPARSE, None, 0, None, 0,
        ctypes.byref(bytes_returned), None
    )
    f.seek(length-1)
    f.write(b"\0")

Таким образом мы делаем преалокацию файла с возможностью писать в любое место, например так работают торренты.
В моем случае было многопоточное скачивание разных кусков файлов с возможностью докачки.

При копировании таких файлов чаще всего копия занимает всё положенное ей место.
Чтобы учитывать такое свойство файла нужно использовать специальные опции

shutil.copyfile(src, dst, follow_symlinks=False)

rsync -S ...

robocopy /SPARSE ...

Для тестирования трансфера требовалось создавать реальные файлы с рандомными данными. Сделать это просто:

import os
with open(file_path, "wb") as out:
    for _ in range(1024):
        out.write(os.urandom(1024*1024*10))

dd if=/dev/urandom of=file.bin bs=1M count=10

Тут, конечно, никаких разреженных файлов быть не может.

#tricks