эх, вынужден согласиться, что с глобальными переменными бывают проблемы такие, которые перевешивают их преимущества :)

И похоже мне в случае когда у меня активно БД используется в мультипотоках помимо запросов фласка, надо весь код делать в духе:

def some_work():
 with db_connection() as db:
    db.query(…)

В джанге, наверное, потратили много часов, чтобы обеспечить эти глобальные переменные.

@contextmanager
def get_session():
    session = get_the_session_one_way_or_another()

    try:
        yield session
    except:
        session.rollback()
        raise
    else:
        session.commit()

th get_session() as session:
    # do something with the session
,

Тебе это надо делать всего в нескольких местах

По сути в каждом треде один раз

Или вообще выделить из этого мини-фреймворк

Салют всем. Такой вопрос:
Допустим я создаю магазин видеоигр. У модели игра есть колонка категория. И есть перечень этих категорий. Но есть вариант что будет категория экшон и подкатегории(фпс, тпс и подобные). Как это правильнее реализовать? Заранее спасибо

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

Причем часто бывает, что какой-то метод my_service.some_func() может использоваться одновременно как в фласке, так и в бекграундных задачах.

И если уж и решать проблему, так лучше сразу глобально везде. С этим with db_connection() as db: у меня хотя бы будет уж точно полный контроль над БД и сессиями. И тогда точно какая-то проблема в одном таске не отразится ни на что другое.

В принципе не так уж и больше кода придется писать. Это просто джанга научила халяве с глобальными переменными )

Так не. Тебе нужна прослойка как для фласка, которая будет управлять сессией

Во фласке у тебя есть запрос, сделать аналогично "таск"

Внутри таска будут крутиться твои функции, а сам таск будет обслуживать сессии и вызов нужного обработчика

Будет что-то типа

Thread(target=task_with_connection(handler_func, session_maker)).start()

Кто-то делал для SQLAlchemy анотацию типов для работы с сессиями через @contextmanager ?

Вот этот код прекрасно понимает PyCharm, автодополнение кода работает как надо:

@contextmanager
def db_session() -> ContextManager[Session]:
    session: Session = DBSession()
    try:
        yield session
        session.flush()
        session.expunge_all()
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

Но на этот код ругается mypy, который я использую в pre-commit хуках:
error: Argument 1 to "contextmanager" has incompatible type "Callable[[], ContextManager[Any]]"; expected "Callable[..., Iterator[<nothing>]]”

Если же я делаю как хочет mypy: def  db_session() -> Iterator[Session] ,то теперь mypy счастлив, но вот PyCharm больше не делает автодополнение по коду, он перестает понимать, что этот контект менеджер возвращает объекти типа Session.

Как бы сделать так, чтобы PyCharm и mypy договорились бы друг с другом?

P.S. Причер усмирить mypy не так просто. Если в определении db_session я могу успокоить mypy через комменты типа # type:ignore,

то он начинает ругаться на все те места, где я этот контекс сессий начинаю использовать: error: Need type annotation for 'db

@contextmanager
def db_session() -> ContextManager[Session]:
    session: Session = DBSession()
    try:
        yield session
        session.flush()
        session.expunge_all()
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

Какой ужас.

Зачем этот декоратор вообще?

Чтобы я мог бы внутри моих сервисов делать такой код для работы с БД:

def my_func():
  with db_session() as db:
    return db.query(MyModel).all()

Я понимаю что ты хочешь сессию. Посмотри как это фласк алхимия делает например. Но часть с "пробуем закомитить и если произошла любая ошибка то делаем ролбек" ужасает.

Хм, верно, спасибо за указание на это. Этот код скопипастил откуда-то, мне вообще в моем случае не нужен блок обработчика ошибки.

Flask-Alchemy скорее всего использует scoped_session. Это шикарно подходит для работы в духе “запрос-ответ” на самом фласке. Но у меня помимо фласка в том же питонячем процессе крутится много всякого в мультипотоках, что использует БД. И если где-то с БД случается ошибка, то я потом psycopg2.errors.InFailedSqlTransaction, мол надо сделать rollback для транзакции.

И чтобы такого не было, хочу использовать пол коннектов, и на каждый момент работы с БД получать отдельную сессию из пула.

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

Ой. Это я упустил, когда читал

Ну я лет 10 уже так работаю. Все мои проекты это вебсервисы, где веб часть — это самое мелкое, для работы операторов, UI интерфейс. И все оснавная часть проектов делается внутри бекграундных сервисов. Делать работу бекграундных тасков через что-то подобное Celery —  не удобно. У меня часто используются локи на работу с объектами, и надо жить в рамках одного питонячего процесса.

До этого работал с MongoDB, там проблем с транзакциями нет, все шикарно. Еще каким-то магическим образом джанга на моих проектах так же  хорошо работает. Я не знаю как, но она через свои глобальные переменные делает так, что если в одном потоке у меня случилась SQL ошибка, то это не затрагивает другие.

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

Проблема именно в линтере mypy, который не понимает аннотацию типа ContextManager[Session].

Претензия была про "тот же процесс". Масштабировать как?