Оператор 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

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

Все паблики облетела новость о покупке Astral. Мнения бытуют разные, так что мне сложно даже предполагать к чему это приведёт.
Сегодня всё так быстро меняется и происходит то, во что раньше бы никто не поверил! Вобщем, будем надеятся.

https://openai.com/index/openai-to-acquire-astral/

#offtop

Мы используем Makefile думая, что нет альтернатив, что это стандарт и всё такое.
Но make это не запускалка команд, а система сборки. Мы фактически используем его не по назначению.
И на самом деле альтернатива есть! Некоторое время назад я открыл для себя прекрасный инструмент - just. Он решает все проблемы make.

just - это не система сборки как make, это именно исполнитель команд!
Больше никаких Phony Targets и табуляций, привет нормальный синтаксис и передача аргументов!!! 😎

⭐️ Что умеет just:

✅ Автодокументирование команд
Не нужно делать отдельную команду с докой, просто добавь комментарий

# команда сборки
build:
    ...

$ just --list
Available recipes:
    build # команда сборки

Команда с именем default запускается по умолчанию если не указано другое, так что я обычно делаю так:

default:
    just --list

Теперь просто выполняем just и получаем доку из текущего файла.

✅ Удобная работа с переменными окружения

# загрузить из .env
set dotenv-load

# глобальная переменная
export PYTHONPATH := "./src"

# переменная для команды
test $TESTUNG="true":
    pytest

✅ Передача аргументов

build target:
    @echo 'Build {{target}}...'

команда запуска

$ just build dev
# Build dev...

✅ Выбор интерпретатора прямо в команде
Пример с инлайн-скриптом на python:

system:
    #!/usr/bin/env python3
    import platform
    print(platform.system())

Эта же функция позволит выполнить скрипт как одну команду вместо перезапуска шела для каждой строки

foo:
  #!/usr/bin/env sh
  for file in ls .; do
    echo $file
  done

✅ Выполнение команды в определенной директории. Можно указать как релятивный путь так и абсолютный

[working-directory: 'backend']
build:
    docker compose build

Также можно задать рабочую директорию глобально

Там еще много интересного:
- поддержка функций
- автокомплиты и интеграции
- экспрешены
- алиасы команд
- группировка команд
- альтернативы команды под разные ОС
- импорт других just-файлов
- цветной вывод
- ... и другие штуковины!

Так что вперёд - ➡️ читать доку!
Репозиторий: ➡️ https://github.com/casey/just
Статья: ➡️ https://www.chicks.net/reference/file_formats/just/

ЗЫ. Кажется, на Makefile я уже не вернусь)

#tools

Если запустить 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

Сгеодня перестал работать Telegram в РФ. Жаль, что вы уже не прочитаете это сообщение. Но если вдруг прочитаете то знайте - канал переехал в MAX!
Подписывайтесь чтобы быть на связи 😎

Стандартная библиотека asyncio это стандарт (начиная с Py3.4) для работы с асинхронным кодом. Но эта библиотека достаточно низкоуровневая, со своими проблемами, устаревшими подходами.
Чтобы исправить это, были созданы разные обертки и альтернативы с реализацией популярных инструментов и паттернов асинхронного программирования. Это такие библиотеки как:

- trio: улучшает корректность выполнения, не оставляя потерянных корутин при ошибках, то есть предлагает Structured Concurrency из коробки.

- curio: упрощение синтаксиса и читаемости кода, больше похоже на работу с потоками.

- anyio: универсальная обертка над asyncio или trio плюс множество вспомогательных инструментов.

anyio используется в FastAPI как основная библиотека для работы с асинхронным кодом и вызовом синхронного кода из асинхронного.

В общем, рекомендую почитать про возможности anyio, возможно вы более не будете использовать чистый asyncio в своих проектах)

Это совсем не значит что дефолтный asyncio плох, он тоже даёт достаточный для работы функционал и продолжает развиваться. Например, в версии 3.11 появились TaskGroup, с похожим на trio функционалом. Так что он тоже актуален, просто придется больше написать кода самостоятельно.

#libs #async

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

from fastapi.concurrency import run_in_threadpool

async def execute(data: DataRequest) -> DataResponse:
    try:
        result = await run_in_threadpool(sync_function, data)
        return DataResponse(data=result)
    except Exception as e:
        return DataResponse(
            error=str(e),
            success=False,
        )

В общем работает нормально. Для всех вызовов под капотом используется общий тредпул, всё работает предсказуемо.
Но потребовалось изменить количество запускаемых в пуле потоков (по умолчанию создается 40 воркеров).
Так как дело происходит с FastAPI, делается это через lifespan используя настройки anyio:

import anyio

@asynccontextmanager
async def lifespan(app: FastAPI):
    limiter = anyio.to_thread.current_default_thread_limiter()
    limiter.total_tokens = 100
    yield
    # если вдруг нужно вернуть обратно
    limiter.total_tokens = 40

Зачем менять количество воркеров?
- уменьшить, если оперативки мало (один тред занимает ~8мб)
- увеличить чтобы выдержать нагрузку

Если есть предложения получше при тех же вводных - предлагайте😉

#async

Nuitka 4.0

Библиотека для компиляции python-кода в исполняемые файлы получила мажорный апдейт.

Ключевые изменения:
- ускорение сборки бинарника в 15 раз
- экспериментальная поддержка компилятора Zig
- возможность выбранные функции оставлять как есть, в виде байт кода через декоратор @nuitka_ignore
- бинарники работают до 30% быстрей (CPU-bound задачи)
- теперь можно вместо отдельного скрипта сборки использовать pyproject.toml. Весь конфиг в одном файле!
- улучшен контроль подключения DLL библиотек для Windows
- улучшена совместимость с рядом популярных и не очень пакетами

И в целом выбран путь на повышение производительности бинарной сборки.

Полная информация ➡️ здесь.

#libs

В асинхронных приложениях есть один не всегда очевидный момент, который приводит к неявным багам - это общие глобальные объекты. Да, это в целом антипаттерн, но иногда такие объекты действительно нужны и вполне уместны. Например, когда вы завязаны на уже существующем коде и не можете его поменять, но переменную подставлять надо, при этом явно пробросить её не получится. Для этого используется ContexctVar.
Случаи бывают разные но я приведу самый понятный пример - логирование.
Допустим, мы не можем передать id юзера куда-то внутрь фреймворка, но можем использовать его в своем хендлере подставляя как переменную. В примере будет без хендлера но суть та же.

import asyncio
from contextvars import ContextVar
import random

# это переменная, которая будет разной для каждой корутины
user_id_ctx = ContextVar("user_id", default=-1)

async def handle_request(user_name, user_id):
    # устанавливаем значение для текущей корутины и её дочерних корутин
    user_id_ctx.set(user_id)
    print(f"Create {user_name} == {user_id}")
    await asyncio.sleep(random.random())
    # id не передаётся в вызов
    await process_order(user_name)

async def process_order(user_name):
    # получаем значение из локального контекста
    current_user_id = user_id_ctx.get()
    print(f"Log {user_name}: {current_user_id}")

async def main():
    await asyncio.gather(
        *[handle_request(f"user {i}", i) for i in range(10)],
    )

if __name__ == "__main__":
    asyncio.run(main())

Во всех выводах id должен совпадать.

- Точно так же можно подставлять атрибуты у инстансов синглтонов.
- Контекст наследуется дочерними корутинами.
- Не стоит увлекаться этим способом, он прилично усложняет логику. Явное лучше неявного.

@asyncio @tricks

Теперь аналогичная история с тредами. Для тредов используется объект 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

В Python 3.6 был полностью переработан стандартный dict. Вместо разреженной таблицы данные стали храниться в плотных массивах. Это дало буст к скорости и экономию памяти. И, как сайд эффект, ключи стали упорядочены. В каком порядке добавляешь ключи, в таком можно и забрать.

Но при этом OrderedDict никуда не делся. Это сделано для совместимости?

Нет, между dict и OrderedDict всё ещё большая разница.

▫️ При сравнении в обычном dict проверяется только наличие ключа, а в OrderedDict проверяется их порядок
▫️OrderedDict основан на связном списке и имеет метод move_to_end() для изменения порядка элементов.
А метод popitem() позволяет удалять элемент как с конца так и из начала.
▫️OrderedDict это кастомный класс. Он не так оптимизирован как обычный dict. Работает дольше, места занимает больше.

В версии 3.7 он был переписан на С и стал быстрей, но всё еще уступает обычному dict.

Немного тестов:

Память: в 2.5-3 раза больше
Создание: в ~2 раза дольше
Удаление: в ~3 раза быстрей (popitem против del)
Поиск по ключу: примерно одинаково (хеш таблицы)

Код тестов↗️

Если вы используете OrderedDict, то это предполагает, что порядок ключей важен для логики программы.

#libs

Multi Tenancy - это архитектурный подход в разработке, при котором один экземпляр приложения обслуживает множество разных клиентов так, что они имеют доступ только строго к своим данным. Например, сервис для обслуживания компании, ведения проектов, изолированных воркспейсов и тд. Наверняка вы использовали такие в работе. От простого почтового сервера до CMS.

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

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

Минусы:
❌ Очевидно - сложность реализации.
Плюс разные особенности разных подходов к решению.

Способ задать контекст, может быть реализован по-разному. Например, в Django есть приложение django.contrib.sites которое реализует разделение через домены. Это самая популярная техника.

▫️ Разделение по домену
https://company1.my-app.com - домен для одной компании
https://company2.my-app.com - домен для второй компании

▫️ Разделение по url
https://my-app.com/client1/
https://my-app.com/client2/

▫️ Определение ID на уровне хидера, выдаётся в момент аутентификации

GET /api/v1/users HTTP/1.1
Host: api.myapp.com
X-Tenant-ID: client-id

▫️ Сохранение в JWT, то есть сохраняется в токен как кастомный payload.


Способ изоляции данных также имеет несколько решений.
▫️ разные БД
▫️ разные схемы в БД (не все СУБД поддерживают)
▫️ по FK-полям в общей схеме

Каждый подход к разделению данных тоже имеет свои плюсы и минусы.

Работает логика так: каждый запрос юзера на ранних этапах определяет контекст (в какой компании, проекте, воркспейсе находится юзер) и это влияет на то, какие настройки он подтягивает, как фильтруются запросы в БД.

В следующем посте рассмотрим один практический пример на базе FastAPI

#systemdesign

Недавно я реализовал Multi Tenancy систему на FastAPI + SQLAlchemy и пересобрал её в небольшой демо проект.

Ключевые особенности:
▫️ Основной всего является сущность Компании. Юзеры могут подключаться к разным компаниям, но в один момент времени могут находиться только в одной (по условиям ТЗ).
▫️ Разделение данных через FK-ссылку всех ключевых бизнес-сущностей на ID компании.
▫️ ID компании задаётся при логине и хранится в JWT. При смене компании потребуется новый токен.
▫️ Контекст активируется через зависимость.
▫️ Фильтр применяется автоматически для любого запроса.

Теперь подробней:

1️⃣ Логин
Когда юзер логинится, он указывает ID компании. Этот ID зашивается в JWT.

2️⃣ Запрос
Ключевое действие находится в функции get_current_user(...). Мы парсим JWT и достаем инфу о юзере и компании.
Для текущей и дочерних корутин этого запроса объявляется контекст конкретной компании. Объявление контекста происходит через ContextVar (а вот и реальный пример их применения).
Никакого указания ID проекта\магазина\сайта со стороны юзера и проброса его по всем функциям. Это локальное состояние конкретного запроса и его цепочки корутин.

3️⃣ Активация фильтра для SQL
Реализация фильтра сделана штатными средствами SQLAlchemy - прослушивание ивента do_orm_execute и правка любого SQL запроса с with_loader_criteria. Фильтр реагирует только на модели с миксином CompanyMixin. Нужно только указать ID компании для филтра.
Функция фильтрации add_tenant_filter(...) требует либо указать контекст, либо отключить фильтр. Если вдруг контекст не задан, то по умолчанию юзер просто получит ошибку BadRequestError.

4️⃣ Админка
В get_current_user() указание контекста необязательно. Это сделано для того, чтобы мог залогиниться суперадмин. Ведь должен же кто-то управлять компаниями "сверху".

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

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

Так же есть один юзерский запрос с bypass_company_filter() - это получение всех доступных юзеру компаний чтобы можно было на UI выбирать куда переключиться.

Если же админу нужно вызвать юзерский сервис с контекстом какой-либо компании, то есть функция set_company_context(...), которая временно включает фильтр и активирует ID указанной компании. Например получить все [имя сущности] юзера для конкретной компании.

▫️Плюсы такого подхода:
- Контекст хранится в токене, нет лишнего запроса в БД
- При утечке токена одной компании другие не пострадают
- Фильтр автоматически применяется на любой запрос юзера
- Все данные в одной БД. Можно связывать данные разных клиентов через FK и хранить общие сущности без проблем.

▫️Минусы:
- Сломался один клиент - сломались все!
- Нужно продумывать индексы с учётом особенностей multi-tenancy
- Если не следить за чистотой кода и правильностью архитектуры, то можно серьезно накосячить. Правильные тесты - наше ВСЁ!
- Общие миграции

▫️Что еще можно добавить
- партиции таблиц с разделением по полю company_id
- автоматическое добавление текущей company_id для создаваемых юзером объектов. Моя версия в фукнции add_company_id(), но не факт, что это лучший способ.

▫️Как запустить
1. клонируем проект
2. запускаем just run
3. готово!

▫️Как протестить
1. клонируем проект
2. запускаем just test
3. готово!

- Почему just?
- Патамушта!

И главное: мой пример не является эталоном и инструкцией к применению, скорей это эксперимент. Буду рад замечаниям и указаниям на проблемы такого подхода.

Другие статьи на почитать:
↗️один
↗️два
↗️три

#tricks #systemdesign

Модуль pwd может использоваться для считывания информации о пользователях из базы данных паролей Unix (обычно /etc/passwd ).

Получить объект текущего юзера можно так:

import pwd

user_pwd = pwd.getpwnam(username)

Теперь нам доступны основные данные из файла passwd

print('ID:', f'{user_pwd.pw_uid}:{user_pwd.pw_gid}')
print('HOME:', user_pwd.pw_dir)
print('Shell:', user_pwd.pw_shell)

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

▫️ Проверка наличия юзера в системе по имени

def user_exists(username):
    try:
        pwd.getpwnam(username)
        return True
    except KeyError:
        return False

▫️ Преобразование имени юзера в ID
Пример с понижением привелегий процесса когда он запущен от root

import os
import pwd

safe_user = 'nobody'
# находим UID для безопасного пользователя
target_uid = pwd.getpwnam(safe_user).pw_uid
target_gid = pwd.getpwnam(safe_user).pw_gid

# меняем права процесса
os.setgid(target_gid)
os.setuid(target_uid)

# Выполняем какие-то действия
os.chdir('/tmp')
with open('nobody-file.txt', 'w') as f:
    pass

# теперь проверьте, файл будет создан от имени юзера nobody

▫️Преобразование ID юзера в имя

user_name = pwd.getpwuid(user_id).pw_name

▫️Определение домашней директории не текущего юзера

pwd.getpwnam(username).pw_dir

Работает только в Unix-системах

#libs

В прошлый раз был пример с понижением привелегий процесса. Проблема в том, что обратно поднять привилегии нельзя.
Чтобы не потерять уровень доступа текущего процесса, нужно выполнять понижение в дочернем процессе. Для этого можно использовать os.fork().

# example.py
from pathlib import Path
import os
import sys
import pwd

def example(username):
    # получаем pwd юзера по имени
    try:
        pw_record = pwd.getpwnam(username)
    except KeyError:
        raise Exception(f"User {username} not found")
        return
    # создаем форк процесса
    pid = os.fork()
    if pid == 0:
        # если мы в дочернем процессе - сбрасываем привилегии
        os.setgid(pw_record.pw_gid)
        os.setuid(pw_record.pw_uid)
        # создаем файл от имени другого юзера
        Path('user-file.txt').touch()
        sys.exit(0)
    else:
        # родительский процесс, создаем файл от имени исходного юзера
        Path('root-file.txt').touch()
        os.waitpid(pid, 0)

example('paul')
# проверяем владельца файлов
print('user-file.txt', Path('user-file.txt').owner())
print('root-file.txt', Path('root-file.txt').owner())

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

~$ sudo python example.py
user-file.txt paul
root-file.txt root

PS. В дочернем процессе os.fork() вернёт 0, чтобы можно было понять в каком процессе продолжает исполняться код. В родительском это будет реальный PID. Это написано в документации.

# tricks

У меня основная ОС это Linux. Но нередко требуется делать сборку клиента под Windows. Собираю обычно через Pynstaller/Nuitka.
И вот этапы борьбы поиска удобного решения.

▫️ Dual Boot
Пока был дуалбут, я перезагружался в Windows, копировал исходники и запускал сборку. Рабочий вариант, но времени занимает слишком уж много. К тому же дуалбута давно нет.

▫️ VirtualBox
Если нет дуалбута, то выручает VirtualBox. Тут стало проще, я расшарил папку с проектом в Windows как сетевой диск. Просто запускаю готовый скрипт сборки и готово. Файл сохраняется сразу на место. Но горький опыт заставил делать шару ReadOnly, что требует сначала копировать проект на локальный диск в виртуалку и только потом запускать сборку. После чего перебрасывать результат. В общем, тоже так себе вариант 😖

Ясное дело, что эти два способа достаточно наивны и никакой автоматики. Поэтому пошли далее...

▫️ Wine
Выглядит всё просто. Выполняем обычные виндовые команды в терминале, просто в начале нужно дописывать wine.
Попробуйте выполнить команду wine cmd и вы окажетесь в "обычной" виндовой консоли. Ну а там просто выполняем команды сборки.
Главное - установить все необходимые зависимости.

В целом запускается, но дебажить не удобно. Проверить запуск на "чистой винде" тоже не получится, только через тот же Wine. Поэтому в итоге я остановился на следующем варианте.

▫️ VirtualBox CLI
Это тоже самое что пункт 2 но полностью на автомате. Используя команду VBoxManage можно манипулировать виртуалкой.

- VBoxManage startvm - запускает виртуалку по имени, в том числе headless
- VBoxManage guestcontrol "VM_NAME" "CMD" позволяет запускать шел-команды
- copyto и copyfrom - позволяет копировать туда-сюда файлы

В результате, мы можем кодом запустить виртуалку, выполнить необходимые действия, забрать результат и погасить виртуалку!

В скрипте я запускаю VM, копирую в неë батник сборки и исходники и запускаю. После сборки файл копируется обратно на хост в dest.

А если надо, запускаем виртуалку в обычном режиме и визуально дебажим скрипты сборки и само приложение.
Из минусов только 20 занятых гигов на винду.

Итого, получился такой демо-проект↗️ со сборкой простого диалога на PySide6. В проекте примеры с wine и VirtualBox CLI

Не обещаю, что у вас запустится сразу, но у скрипты точно рабочие!

Какие еще есть варианты

▫️В Docker контейнере. Например есть такой cdrx/pyinstaller-windows
▫️Говорят PyOxidizer в будущем сможет делать это без танцев с бубном, но пока не реализовано. Только вот проект 2 года не обновляется, может и не реализуют уже 😢

#pyside #linux

Не так давно столкнулся с API который в ответе с ошибкой присылал только статус-код. Без подробностей и текста ошибки.
Неудобно когда нет явного описания что произошло, но вот такой сервис достался 😐

Чтобы сообщение совсем не было пустым хотелось бы показывать хоть что-то, например стандартный текст ошибок соответствующего кода.
В стандартных библиотеках текст статус-кодов можно достать из модуля http.

from http import HTTPStatus

HTTPStatus(404).phrase
# 'Not Found'
HTTPStatus(201).phrase
# 'Created'
HTTPStatus(418).phrase
# "I'm a Teapot"

#libs 🫖