Python Заметки

Все паблики облетела новость о покупке Astral. Мнения бытуют разные, так что мне сложно даже предполагать к чему это приведёт. Сегодня всё так быстро меняется и происходит то, во что раньше бы никто не поверил! Вобщем, будем надеятся. https://openai.com/index/openai-to-acquire-astral/ #offtop

Еще одно применение пайпов - в контексте с 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

Оператор pipe позволяет писать более компактный код, реализуя логику объединения данных (Union). Важно помнить, что его поведение зависит от контекста. Побитовые операции (логическое OR)

result = 5 | 3
# 5 (0101) | 3 (0011) = 7 (0111)

Самое главное - не путать с оператором or, это другое! Объединение множеств

set_a = {1, 2, 3}
set_b = {3, 4, 5}
set_c = set_a | set_b  
# {1, 2, 3, 4, 5}
set_c |= {5, 6}
# {1, 2, 3, 4, 5, 6}

Слияние словарей

dict_1 = {"a": 1, "b": 2}
dict_2 = {"b": 3, "c": 4}
merged = dict_1 | dict_2  
# {'a': 1, 'b': 3, 'c': 4}
merged |= {"d": 5}
# {'a': 1, 'b': 3, 'c': 4, 'd': 5}

Аннотации типов, заменяет Union

def process_data(value: int | str) -> None:
    print(value)

Допустимо использовать в isinstance или issubclass

isinstance(3, int | float)
# True

Паттерн-матчинг

status_code = 404
match status_code:
    case 200 | 201 | 204:
        print("OK")
    case 400 | 404 | 500:
        print("ERROR")

Для использования в своих классах требуется переопределить метод __or__

Так же нашел библиотеку pipe которая добавляет еще много возможностей. Рекомендую ознакомиться ;) #basic

💐❤️✨🌼⭐️

Почему в прошлом посте я использовал 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

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

@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

А что происходит на противоположном фронте? Вы, вероятно, слышали, что 2026 год называют годом Linux на десктопе (в каких-то узких кругах - годом гейминга на Linux). Всё потому, что экосистема Linux постепенно становится более дружелюбной для обычных десктоп-юзеров (в том числе привыкших к Windows), и не только! ▫️ всё чаще появляются Linux дистрибутивы визуально похожие на Windows (или даже лучше), и множество видео с советами какой дистрибутив попробовать новичкам. ▫️ обновления ядра и любых пакетов в экосистеме Linux всегда привносят оптимизацию и удобство и поддержку свежего железа (привет винде с её обратной тенденцией). Например грядущая версия 7.0, опять с множеством приятных мелочей. ▫️ после 10 лет с последего релиза версии 5 окружение KDE Plasma получила мажорный апдейт версии 6 и активно развивается (уже доросла до 6.5). GNOME тоже не спит и готовит версию 50. ▫️ в Wine добавили патч позволяющий устанавливать продукты Adobe на Linux. Для кого-то это был последний рубеж?😏 ▫️ Proton активно развивается, да так, что через эту прослойку игры работают даже быстрей чем нативно на винде. ▫️ с каждым релизом Wine и Proton поддерживается всё больше игр, что можно отслеживать на ProtonDB, и даже случаются бусты производительности. ▫️ Я сам на днях на виндобук поставил ChacyOS после чего игры, которые тянули гдето в 5-10 FPS, стали играбельными! Подтверждено личным опытом! Кстати, есть несколько дистрибутивов заточенные именно под игры. ▫️ Valve выпускают новую пачку железок которые (предположительно) порвут рынок гейминга (как и в прошлый раз) и (определнно точно) работают на Linux. Именно Valve вливает ресурсы в Linux в целом и в Proton в частности. ▫️ Госсектор разных стран давно уже мигрирует на opensource, так как нет доверия системе которая может одномоментно неконтролируемо массово рухнуть или быть удаленно заблокированной (в том числе по политическим причинам). И ниже небольшой опрос - какая у вас операционка основная?

Ни к чему не призываю, ничего не советую! Просто подмечаю тенденцию и хочется узнать мнения из первых рук 😉

Знаю, что Linux тоже не идеален, знаю что каждой задаче - свой инструмент. Но это не тема поста, так что можно без холиваров)

#offtop #linux

Windows 11 становится всё менее дружелюбна к юзерам а порой и вовсе не юзабельной: ▫️ постоянные ломающие апдейты которые не дают загрузить систему, откатить ломающие апдейты, и давно уже удаляют файлы пользователей без спроса. Тенденция в целом уже достаточно давно, включая глобальные сбои и другие неприятности. ▫️ навязчивое продвижение AI шпионов агентов повсюду в системе которых никто не просил. ▫️ всё больше ресурсов ВАШЕГО компа работают не для вас, а в угоду Microsoft. Мелкомягкие официально предлагают купить железо помощней (чтобы они и дальше могли половину мощности использовать по своему усмотрению) а оно что-то не покупается. Рядовой юзер не понимает зачем менять комп который и так норм работает. А глядя на текущие цены на память наступает ощущение что с этим миром что-то не так. ▫️ люди булшитят винду и активно продвигают переход на Linux порой называя винду кучей слопа или даже вирусом, похищающим файлы с целью выкупа (они реально после аплоада и удаления с локала отключают доступ к файлам и требуют купить подписку). А сам Microsoft переименован в Microslop. Появляются даже тулзы для очистки системы от этого слопа. ▫️ Microsoft уже не скрывает, что ваши данные уже не ваши, даже зашифрованные, ибо ваши пароли давно уже хранятся где надо и доступны кому надо. ▫️ При всех этих факапах они закрыли поддержку Windows 10 не давая возможности откатиться на что-то более стабильное. То есть сами Microsoft стали катализатором поиска альтернатив. Сам я уже на Linux уже более 7 лет как на основной системе, дома винда есть только в виртуалке для тестов клиентского софта. Расскажите, как у вас обстоят дела на винде? Вы пользуетесь системой или боретесь с ней? #offtop

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

Потоковая обработка часто встречается при работе с большими файлами или когда данные приходят частями. В 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

В Gunicorn добавили ASGI. Пока что beta.

gunicorn myapp:app --worker-class asgi

То есть теперь вместо "gunicorn воркеры + uvicorn ранеры" будет всё из одной библиотеки? Коненчо, если скорость не будет драматически ниже. #libs

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

_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

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

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

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

В работе с медиа файлами часто требуется определить не просто расширение, а его, скажем так, "категорию". Тоесть определить это видео, аудио или картинка. Примерно в 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

С Новым Годом! 🎄☃️❄️ Снова этот рубеж подведения итогов и определения планов на следующее 365 дней. Что же мы успели застать в 2к25? 🔸 AI снова делает скачёк в развитии, как по качеству, так и по затратам на ресурсы Продолжая расшатывать все рынки 🔸 Эпичный прорыв цен на железо (из-за первого факта). Сначала оперативка, потом и остальные подтянулись. Успели закупиться вовремя? 🔸 Новый виток "борьбы с интернетом" в РФ Работать всё сложней 🔸 Опенсорсный проект MinIO закрылся Теперь только в облаке и только за денежку Но не всё так плохо! 🔸 Всё больше уверенных мнений, что AI нас не заменит Но всем нужно адаптироваться к новым реалиям и инструментам 🔸 uv ворвался в прод Так и стандартом станет скоро 🔸 Вышел Django 6 Достаточно ли изменений для мажорной версии? 🔸 Вышел PIthon 3.14 с NO-GIL режимом Раньше это считалось невозможным! 🔸 В том же 3.14 мы получили полноценные Субинтерпретаторы и JIT И другие оптимизации 🔸 Язык Rust теперь официально второй язык ядра Linux Хоть и не без проблем 🔸 Проекту pythonotes 6 лет 🎂 Скоро в школу) Мир вокруг меняется постоянно и всё с большей скоростью. Не ищите виновных, просто адаптируйтесь и постоянно учитесь. И всё будет пучком! 😎

Оглядываясь назад в прошлое, задумайтесь, можете ли вы сказать тому себе из прошлого СПАСИБО за то, что вы имеете в настоящем? И хорошенько подумайте в этом настоящем, что нужно делать уже сейчас, чтобы вы из будущего стали лучшей версией себя настоящего и гордились своей версией из прошлого за заботу о будущем. Sir Christopher Edward Nolan :)

#offtop

Отдельно разберём TaskGroup, который пришел на замену gather в Python 3.11. Ключевые отличия ▫️create_task() возвращает объект asyncio.Task, у которого есть соответствюущие методы управления. То есть у нас больше контроля ▫️это контекстный менеджер, который гарантирует что все таски будут остановлены по выходу из контекста ▫️ошибка автоматически отменяет незавершенные задачи, ▫️except* передает нам ExceptionGroup, в котором каждую ошибку можно обработать отдельно

import asyncio
import random

async def do_it() -> str:
    if random.random() < 0.1:
        raise ValueError('Oops')
    delay = random.uniform(0.5, 1.5)
    await asyncio.sleep(delay)
    return delay

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
          for _ in range(10):
              tasks.append(tg.create_task(do_it()))
        for t in tasks:
            print(t.result())
    except *ValueError as e:
        for err in e.exceptions:
            print(err)

asyncio.run(main())

Рекомендую изучить страницу Coroutines and Tasks из документации, где представлено больше интересных примеров и механизмов - таймауты - отмена задач - создание задач из другого потока #async

Функция asyncio.wait() это еще один способ вызвать множество асинхронных задач. Она работает в нескольких режимах. 1. Самый простой - ждем завершения всех задач

async def main():
    tasks = [asyncio.create_task(do_it(i)) for i in range(10)]
    done, pending = await asyncio.wait(
        tasks,
        return_when=asyncio.ALL_COMPLETED
    )
    for task in done:
        try:
            print(task.result())
        except Exception as e:
            print(e)

Очень похоже на gather, но работает не так. ▫️возвращает не результаты, а два сета с объектами Task у которых можно забрать результат через task.result() если они в списке done ▫️не гарантирует порядок результатов так как оба объекта это set ▫️не выбрасывает исключение когда оно появляется, а сохраняет его в Task. Исключение появится когда попробуете забрать резултьтат. 2. Ждем завершения первой задачи, даже если там ошибка.

async def main():
    tasks = [asyncio.create_task(do_it(i)) for i in range(3)]
    done, pending = await asyncio.wait(
        tasks,
        return_when=asyncio.FIRST_COMPLETED
    )
    # в done может быть несколько задач!
    for task in done:
        try:
            print(task.result())
        except Exception as e:
            print(f"Fail: {e}")
    # Оставшиеся задачи в pending, как правило, нужно отменить, иначе они будут продолжать работать
    for task in pending:
        task.cancel()

В сете done будут таски которые успели завершится, причем как успешно так и нет. 3. До первой ошибки. Тоже самое, но с аргументом FIRST_EXCEPTION

done, pending = await asyncio.wait(
    tasks,
    return_when=asyncio.FIRST_EXCEPTION
)

Функция завершается как только первая задача упадет с ошибкой.

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

↗️ Полный листинг примеров здесь #async