Multiprocessing вызвать ошибку внутри process?

Question

[r1dddy4sv]

Можно ли как то остановить Process, вызвав ошибку?
Дело в том, что по окончанию работы сессии multiprocess.Process, я бы хотел чтобы выполнялось действие, показанное в finally:

def drop(args):
    try:
         # do something
         driver=**anything**
         #do something 
    except:
          pass
    finally:
         driver.do_anything()

На данный момент я останавливаю работу сессии следующим образом:

s.terminate()
s.join() 
#Здесь должно выполнится driver.do_anything()

Можно ли как то вызвать ошибку внутри сессии, что бы условие finally или except было выполнено?

Спасибо

Answer 1

[LXSTVAYNE]

Можешь написать кастомный процесс с использованием контекстного менеджера.
Метод __enter__() выполняется на старте, а __exit__() всегда на выходе из контекстного менеджера, ты можешь отловить в нём свои ошибки и сделать необходимые действия.

from multiprocessing import Process


class CustomProcess(Process):
    def __init__(self):
        super().__init__()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(exc_type, exc_val, exc_tb)
        print("Посмертное действие")


if __name__ == '__main__':
    with CustomProcess() as process:
        raise Exception("Some Value")

Comments

[r1dddy4sv]

Благодарю за совет, но у меня появился вопрос, получится ли в методе __exit__() взять переменную из основной функции?
То есть:

from multiprocessing import Process
import time

class CustomProcess(Process):
    def __init__(self):
        super().__init__()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(exc_type, exc_val, exc_tb)
        driver.do_something

def drop():
    driver=**anything**

if __name__ == '__main__':
    with CustomProcess() as process:
        s=process(target = drop, args = (self.lineEdit.text(),proxy,m,p,ii))
        s.start()
        time.sleep(10)
        s.terminate()
        s.join()

[LXSTVAYNE]

r1dddy4sv, ты можешь свою функцию прописать в теле процесса, а переменную сделать полем класса.

[r1dddy4sv]

LXSTVAYNE, Не могли бы вы пожалуйста расписать как это будет выглядеть в коде?

[LXSTVAYNE]

r1dddy4sv,

from multiprocessing import Process


class CustomProcess(Process):
    def __init__(self):
        super().__init__()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(exc_type, exc_val, exc_tb)
        print("Посмертное действие")
        if exc_type is IndexError:  # Можно свой кастомный класс написать
            print(self.my_value)
            return self  # Таким образом мы не выбросим никакую ошибку из контекстного менеджера

    def do_anything(self):
        ...
        self.my_value = 'SOME VALUE'
        ...
        raise IndexError()


if __name__ == '__main__':
    with CustomProcess() as process:
        process.do_anything()

    print(process.my_value)

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

Answer 2

[Виктор Голованенко]

Создайте пользовательское исключение и бросьте его в блоке try:

class MyException(Exception):
    pass


def drop(args):
    try:
         # do something
         driver=**anything**
         #do something 
         raise MyException
    except MyException:
          pass
    finally:
         driver.do_anything()

Comments

[r1dddy4sv]

Спасибо за ответ, но хотелось бы вызывать ошибку при остановке сессии, а не в конце функции. То есть что бы пользователь во время исполнения процесса мог остановить его в любой момент и при этом действие driver.do_anything() будет выполнено