О работе с PostgreSQL JSONB в SQLAlchemy

JSON поля в PostgreSQL - это действительно удобный способ хранения данных, наряду с реляционной моделью. Они могут понадобиться там, где нужно произвести денормализацию данных для ускорения работы или для хранения гетерогенных данных.

Алхимия поддерживает работу с JSON-полями нативно, поэтому мы можем добавлять эти поля в свои модельки. Но есть пара нюансов, связанных с тем, каким образом алхимия проверяет изменения в этих полях.

Допустим, есть у нас вот такой класс:

class Config(Base):
id = Column(Integer, primary_key=True)
config = Column(JSONB)
tablename = 'config'

А теперь попытаемся добавить запись и изменить её:

cfg = Config(
    id=1,
    config={'some_const': 1}
)

session.add(cfg)
session.commit()

cfg.config['test'] = 2

session.add(cfg)
session.commit()

# упадёт с assertion error
assert cfg.config['test'] == 2

Почему оно упадёт с assertion error?
Внимательно прочитаем документацию SQLAlchemy и найдем вот такой забавный пункт: The JSON type, when used with the SQLAlchemy ORM, does not detect in-place mutations to the structure. То есть, при изменения структуры нашего объекта Алхимия не понимает, произошло ли какое-то изменение или нет.
Есть 2 решения этой проблемы:

1) Использовать класс MutableDict, если у нас нет вложенности в JSONB объектах:
Просто оборачиваем этой штукой наш тип в модели и радуемся жизни. Выглядит это как-то так:

...
 config = Column(MutableDict.as_mutable(JSONB))

2) Используем библиотеку sqlalchemy-json, если у нас есть вложенность:
В данном случае наше поле будет выглядить как-то так:

from sqlalchemy_json import mutable_json_type
...
config = Column(mutable_json_type(dbtype=JSONB, nested=True))

Ссылочки:
1) Adding mutability to json
2) Mutation tracking
3) Mutation tracking in nested JSON structures using SQLAlchemy

#sqlalchemy #рецепт

Если обычного itertools вам мало, то можно использовать more-itertools.

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

Например, вот так можно разделить список на 3 части:

data = ["first", "second", "third", "fourth", "fifth", "sixth", "seventh"]
[list(l) for l in divide(3, data)]
#  [['first', 'second', 'third'], ['fourth', 'fifth'], ['sixth', 'seventh']]

Вот еще задача. Надо разделить список с элементами по определенному условию. В этом нам поможет bucket:

class Cat:
    pass
class Dog:
    pass
shapes = [Cat(), Dog(), Cat(), Dog(), Cat(), Cat()]
result = more_itertools.bucket(shapes, key=lambda x: type(x))
len(list(result[Cat])) # 4
len(list(result[Dog])) # 2

По итогу кошки и собаки будут разделены по своему типу на 2 генератора.

Решим самую частую проблему - перевести список с несколькими уровнями вложенностями в "плоский" список:

iterable = [(1, 2), ([3, 4], [[5], [6]])]
list(more_itertools.collapse(iterable)) #[1, 2, 3, 4, 5, 6]

А если в плоский список нам нужно вытащить только элементы с первым уровнем вложенности?

list(more_itertools.collapse(iterable, levels=1)) # [1, 2, [3, 4], [[5], [6]]]

А вот так мы можем посмотреть, все ли элементы в коллекции уникальные:

more_itertools.all_unique([1, 2, 3, 4]) # True
more_itertools.all_unique([1, 2, 1, 4]) # False

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

PyPI | Документация

#more_itertools #itertools #библиотека #рецепт

На сегодня расскажу ещё пару рецептов с more_itertools.

1) map_if работает как обычный map, но применяет функцию на элемент только если оно попадает под условие. Например, вот так мы можем возвести в квадрат только те числа, которые делятся на 2 нацело:

example = [1, 2, 3, 4, 5, 6, 7, 8]
list(map_if(example, lambda x: x % 2 == 0, lambda x: x * x)) # [1, 4, 3, 16, 5, 36, 7, 64]

2) Получить последний элемент можно при помощи last. Возникает вопрос а зачем он существует, если можно указать sequence[-1]? Ответом является то, что last позволяет указать, что ему возвращать, если элементов в коллекции нет:

last([1, 2, 3]) # Очевидно получим 3 
last([], 0) # Список пустой, но получим 0
[][-1] # Получим IndexError

Ещё есть first - как понятно из названия, он получает первый элемент.

3) map_except тоже работает как map, но умеет игнорировать ошибки. Например, мы хотим получить только те элементы, которые получилось привести к целому числу:

example = [1, "1", "2", "test", "three", object, 4.0]
list(map_except(int, example, ValueError, TypeError)) # [1, 1, 2, 4]

4) Ну и в конце про take - он просто берет N элементов из итерируемого объекта:

example = range(10)
take(3, example) # [0, 1, 2]
take(20, example) # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] - если больше, возьмет доступные

#itertools #more_itertools #библиотека #рецепт

Там в Python 3.12 добавили нашумевший PEP 659, а у меня пет-проект один давно не обновлялся, и так уж звёзды сошлись, что я сижу второй день обновляю его на 3.12

Задача - есть функционал, который под капотом имеет некоторый класс следующего вида:

class BaseFunction:
    serialize_to: None

    def serialize(self, data: dict) -> serialize_to:
        pass  # тут мы используем наш serialize_to

@dataclass
class ModelA:
    x: str

class FunctionA(BaseFunction):
    serialize_to: ModelA

Мы определяем новые классы наследуясь от BaseFunction, переопределяем в них serialize_to и вызываем serialize который делает нам инстанс serialize_to.

Ну прямо дженерик напрашивается! Тем более в 3.12 их завезли, красивые:

class BaseFunction[T]:
    def serialize(self, data: dict) -> T:
        pass # тут мы используем наш serialize_to

class FunctionA(BaseFunction[ModelA]):
    pass

Встаёт вопрос, а как нам получить наш тип из дженерика?

Для начала, получим __orig_bases__[0] - он вернёт нам классы, от которых мы наследовались. Так как нам нужен только наш первый класс, мы указываем [0]:

>>> FunctionA.__orig_bases__
__main__.BaseFunction[__main__.ModelA]

Ещё можно это сделать с помощью get_original_bases из types, но его добавили только в 3.12 (почему я об этом сказал - узнаете ниже).

Теперь надо получить получить сам тип в дженерике. В этом нам поможет typing.get_args, который получает все аргументы типа. Дополнительно укажем, что нам нужен первый тип:

>>> get_args(FunctionA.__orig_bases__[0])[0]
<class '__main__.ModelA'>

Теперь в методе serialize класса BaseFunction[T] можно написать штуку, которая автоматически сериализует наши данные:

def serialize(self, data: dict) -> T:
    type_from_generic = get_args(self.__class__.__orig_bases__[0])[0]
    return type_from_generic(**data)

Проверяем:

>>> f = FunctionA()
>>> f.serialize(data={"x": 1})
ModelA(x=1)

Вы восхитительны!

Кстати, эта же штука должна работать ещё вроде как аж с 3.8, так как в нём именно был добавлен __orig_bases__ (PEP 560), ну и под капотом у новых дженериков используется...

>>> FunctionA.__mro__
(<class '__main__.FunctionA'>, <class '__main__.BaseFunction'>, <class 'typing.Generic'>, <class 'object'>)

Ага, typing.Generic :)

#рецепт #std