Source code for gavo.utils.monkeypatch

"""
This module contains monkey patches for the python standard library
where we want to use features beyond the current base line (python 3.7).

It is imported early on by utils.__init__, so anything that is in DaCHS
should be able to rely on these patches.
"""

#c Copyright 2008-2023, the GAVO project <gavo@ari.uni-heidelberg.de>
#c
#c This program is free software, covered by the GNU GPL.  See the
#c COPYING file in the source distribution.


import functools
import threading

if not hasattr(functools, "cached_property"): # pragma: no cover
	functools._NOT_FOUND = object()

	class cached_property:
		def __init__(self, func):
			self.func = func
			self.attrname = None
			self.__doc__ = func.__doc__
			self.lock = threading.RLock()

		def __set_name__(self, owner, name):
			if self.attrname is None:
				self.attrname = name
			elif name != self.attrname:
				raise TypeError(
					"Cannot assign the same cached_property to two different names "
					f"({self.attrname!r} and {name!r})."
				)

		def __get__(self, instance, owner=None):
			if instance is None:
				return self
			if self.attrname is None:
				raise TypeError(
					"Cannot use cached_property instance without calling __set_name__ on it.")
			try:
				cache = instance.__dict__
			except AttributeError:  # not all objects have __dict__ (e.g. class defines slots)
				msg = (
					f"No '__dict__' attribute on {type(instance).__name__!r} "
					f"instance to cache {self.attrname!r} property."
				)
				raise TypeError(msg) from None
			val = cache.get(self.attrname, functools._NOT_FOUND)
			if val is functools._NOT_FOUND:
				with self.lock:
					# check if another thread filled cache while we awaited lock
					val = cache.get(self.attrname, functools._NOT_FOUND)
					if val is functools._NOT_FOUND:
						val = self.func(instance)
						try:
							cache[self.attrname] = val
						except TypeError:
							msg = (
								f"The '__dict__' attribute on {type(instance).__name__!r} instance "
								f"does not support item assignment for caching {self.attrname!r} property."
							)
							raise TypeError(msg) from None
			return val

	functools.cached_property = cached_property


# Work around a rather severe openssl bug leading to presumably
# exploitable crashes https://twistedmatrix.com/trac/ticket/9764
from OpenSSL import SSL

def patch_openssl():
	original = SSL.Context.__init__

	def newInit(self, method):
		original(self, method)
		self.set_session_cache_mode(SSL.SESS_CACHE_OFF)
	
	SSL.Context.__init__ = newInit

patch_openssl()