Spaces:

sisithefox
/

sisi-astrology-api

Running

sisi-astrology-api / calculation_lock.py

Serialize shared Western and Vedic calculation state

436ab72 20 days ago

779 Bytes

"""Serialize calls to process-global Swiss Ephemeris/PyJHora state.
PyJHora changes shared ayanamsa/node settings and patches swisseph functions.
Its temporary stdout replacement is process-global too. FastAPI runs ordinary
def routes in a thread pool, so all chart entry points must use the same lock.
Each server process owns its own library state and lock.
"""
from functools import wraps
from threading import RLock
from typing import Callable, ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")
_calculation_lock = RLock()


def serialize_calculation(function: Callable[P, R]) -> Callable[P, R]:
    @wraps(function)
    def locked(*args: P.args, **kwargs: P.kwargs) -> R:
        with _calculation_lock:
            return function(*args, **kwargs)
    return locked

	"""Serialize calls to process-global Swiss Ephemeris/PyJHora state.

	PyJHora changes shared ayanamsa/node settings and patches swisseph functions.
	Its temporary stdout replacement is process-global too. FastAPI runs ordinary
	def routes in a thread pool, so all chart entry points must use the same lock.
	Each server process owns its own library state and lock.
	"""
	from functools import wraps
	from threading import RLock
	from typing import Callable, ParamSpec, TypeVar

	P = ParamSpec("P")
	R = TypeVar("R")
	_calculation_lock = RLock()


	def serialize_calculation(function: Callable[P, R]) -> Callable[P, R]:
	@wraps(function)
	def locked(args: P.args, *kwargs: P.kwargs) -> R:
	with _calculation_lock:
	return function(args, *kwargs)
	return locked