Написание атомарных функций на Python

В Python существует несколько вариантов создания атомарных функций. Они обычно известны как транзакции и часто требуют импорта.

1. Вы можете использовать sqlite3 и подключить его к своей программе Python3, чтобы начать транзакцию и убедиться, что она завершена.

2. Вы можете использовать несколько блоков Try-кроме-finally, чтобы убедиться, что весь блок завершен, и чтобы он завершился неудачно, если это не так.

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

Вот пример №2:

import os
import shutil

class AtomicOperation:
    def __enter__(self):         
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        if exc_type:
            self.rollback()
        else:
            self.commit()

    def commit(self):

    def rollback(self):

def create_user_record(user_record):
    pass

def create_directory(directory_name):
    os.makedirs(directory_name)

def perform_operations(user_record, directory_name):
    with AtomicOperation() as operation:
        try:
            create_user_record(user_record)
            create_directory(directory_name)
        except Exception as e:
            print(f"Operation failed: {e}")

Транзакции в базах данных легко группировать, поскольку все операции выполняют одно и то же: работают с данными, имеющими четко определенное «предыдущее» состояние, к которому можно выполнить откат.

«Общая» атомарная операция для «всего» не может быть выполнена для действий, которые имеют «побочные эффекты» в системе. Предположим, что в самом вашем примере операции «create_user_record» и «create_directory» выполняются в файловой системе, а «create_directory» завершается с ошибкой? Как среда выполнения Python может узнать, в каком состоянии была файловая система до сбоя, если предполагается, что эти две операции должны быть атомарно сгруппированы?

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

В общем, это возможно, но вам придется создать специальный объект-менеджер, аналог соединения с базой данных, и выполнить все изменения, которые вы хотите сделать атомарными, с помощью этого объекта — либо явно, либо с помощью объекта, неявно защищающего ваши данные. операции с побочными эффектами - и чем больше доменов вы хотите включить в «атомарность» (т. е. файловый ввод-вывод, изменение глобальных переменных, сетевой ввод-вывод с использованием уже созданного конкретного API-клиента), тем более сложным должен быть ваш менеджер. .

Тем не менее, такой объект можно создать, и синтаксис контекстного менеджера Python (блок операторов with) идеально подходит для этого. Вот простой класс, который может управлять доступом к глобальным переменным и «отменять» все изменения глобальных переменных аналогично откату:

import inspect

class GlobalVarsTransaction:
    def __enter__(self):
        self.previous_globals = inspect.currentframe().f_back.f_globals.copy()
    def __exit__(self, exc_type, exc_value, tb):
        if exc_type is None:
            # everything fine, allow all global vars set to persist!
            return
        # "rollback" global variables:
        globals = inspect.currentframe().f_back.f_globals
        new_keys = globals.keys() - self.previous_globals.keys()
        globals.update(self.previous_globals)
        for key in new_keys:
            del globals[key]

Обратите внимание, что это задумано как пример и будет охватывать несколько применений в «реальном мире»: в частности, оно не охватывает изменение структуры данных в глобальной области видимости, то есть если кто-то меняет элементы внутри списка или словаря, привязанного к переменной. в текущей глобальной области, которая не отслеживается в этом менеджере.

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

def test():
    global a, b, c, d, e, f, g
    a = b = c = d = e = f = g = 0
    del g

    with GlobalVarsTransaction():
        a = 1
        b = 2
    assert a == 1 and b == 2
    try:
        c = 3
        raise RuntimeError()
        d = 4
    except RuntimeError:
        pass
    # partial state change took place
    assert c == 3 and d == 0
    try:
        with GlobalVarsTransaction():
            e = 5
            raise RuntimeError()
            f = 6
    except RuntimeError:
        pass
    assert e == 0 and f == 0
    try:
        with GlobalVarsTransaction():
            g = 7
            raise RuntimeError()
    except RuntimeError:
        pass
    try:
        g # should fail, as "g" is deleted in the beggining of the test
    except NameError:
        pass
    else:
        assert False, "variable created in failed transactions is set"
test()

Учитывая все вышесказанное, интересно отметить, что если кто-то решит иметь систему, которая может выполнять атомарные операции в разных доменах, было бы хорошей идеей (или даже невозможно без этого) использовать протокол двухфазной фиксации - и сделайте классы Manager, подобные приведенному в этом примере, отдельными для домена, которые будут выполнять необходимые для него шаги.
Если идти в этом направлении, то пакет zope.transactionтранзакций имеет производственное качество, существует уже десятилетия и наверняка поможет правильно его реализовать.