Differential D1690 Diff 9547 pykolab/auth/ldap/auth_cache.py

Changeset View

Standalone View

pykolab/auth/ldap/auth_cache.py

	Show All 11 Lines
	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	# GNU General Public License for more details.			# GNU General Public License for more details.

	# You should have received a copy of the GNU General Public License			# You should have received a copy of the GNU General Public License
	# along with this program. If not, see <http://www.gnu.org/licenses/>.			# along with this program. If not, see <http://www.gnu.org/licenses/>.
	#			#

	import datetime			import datetime
				import os

	import sqlalchemy			import sqlalchemy

	from sqlalchemy import Column			from sqlalchemy import Column
	from sqlalchemy import DateTime			from sqlalchemy import DateTime
	from sqlalchemy import Integer			from sqlalchemy import Integer
	from sqlalchemy import MetaData			from sqlalchemy import MetaData
	from sqlalchemy import String			from sqlalchemy import String
	▲ Show 20 Lines • Show All 51 Lines • ▼ Show 20 Lines


	def del_entry(key):			def del_entry(key):
	# pylint: disable=global-statement			# pylint: disable=global-statement
	global db			global db

	db = init_db()			db = init_db()

				if db is None:
				return

				purge_entries()

	try:			try:
	db.query(Entry).filter_by(key=key).delete()			db.query(Entry).filter_by(key=key).delete()
	except sqlalchemy.exc.OperationalError:			except sqlalchemy.exc.OperationalError:
	db = init_db(reinit=True)			db = init_db(reinit=True)
	except sqlalchemy.exc.InvalidRequest:			except sqlalchemy.exc.InvalidRequest:
	db = init_db(reinit=True)			db = init_db(reinit=True)
	finally:			finally:
	db.query(Entry).filter_by(key=key).delete()			db.query(Entry).filter_by(key=key).delete()

	db.commit()			db.commit()


	def get_entry(key):			def get_entry(key):
	# pylint: disable=global-statement			# pylint: disable=global-statement
	global db			global db

	db = init_db()			db = init_db()

				if db is None:
				return None

	try:			try:
	_entries = db.query(Entry).filter_by(key=key).all()			_entries = db.query(Entry).filter_by(key=key).all()
	except sqlalchemy.exc.OperationalError:			except sqlalchemy.exc.OperationalError:
	db = init_db(reinit=True)			db = init_db(reinit=True)
	except sqlalchemy.exc.InvalidRequest:			except sqlalchemy.exc.InvalidRequest:
	db = init_db(reinit=True)			db = init_db(reinit=True)
	finally:			finally:
	_entries = db.query(Entry).filter_by(key=key).all()			_entries = db.query(Entry).filter_by(key=key).all()

	if len(_entries) != 1:			if len(_entries) != 1:
	return None			return None

	log.debug("Entry found: %r" % (_entries[0].__dict__))			log.debug("Entry found: %r" % (_entries[0].__dict__))
	log.debug("Returning: %r" % (_entries[0].value))			log.debug("Returning: %r" % (_entries[0].value))

	return _entries[0].value.encode('utf-8', 'latin1')			return _entries[0].value.encode('utf-8', 'latin1')


	def set_entry(key, value):			def set_entry(key, value):
				# pylint: disable=global-statement
				global db

	db = init_db()			db = init_db()

				if db is None:
				return

	try:			try:
	_entries = db.query(Entry).filter_by(key=key).all()			_entries = db.query(Entry).filter_by(key=key).all()
	except sqlalchemy.exc.OperationalError:			except sqlalchemy.exc.OperationalError:
	db = init_db(reinit=True)			db = init_db(reinit=True)
	except sqlalchemy.exc.InvalidRequest:			except sqlalchemy.exc.InvalidRequest:
	db = init_db(reinit=True)			db = init_db(reinit=True)
	finally:			finally:
	_entries = db.query(Entry).filter_by(key=key).all()			_entries = db.query(Entry).filter_by(key=key).all()

	if not _entries:			if not _entries:
	db.add(Entry(key, value))			db.add(Entry(key, value))
	db.commit()			db.commit()

	elif len(_entries) == 1:			elif len(_entries) == 1:
	if not isinstance(value, unicode):			if not isinstance(value, unicode):
	value = unicode(value, 'utf-8')			value = unicode(value, 'utf-8')

	if not _entries[0].value == value:			if not _entries[0].value == value:
	_entries[0].value = value			_entries[0].value = value

	_entries[0].last_change = datetime.datetime.now()			_entries[0].last_change = datetime.datetime.now()
	db.commit()			db.commit()


	def purge_entries(db):			def purge_entries():
				# pylint: disable=global-statement
				global db

				try:
	db.query(Entry).filter(			db.query(Entry).filter(
	Entry.last_change <= (datetime.datetime.now() - datetime.timedelta(1))			Entry.last_change <= (datetime.datetime.now() - datetime.timedelta(1))
	).delete()			).delete()

	db.commit()			db.commit()
				except Exception:
				return


	def init_db(reinit=False):			def init_db(reinit=False):
	"""			"""
	Returns a SQLAlchemy Session() instance.			Returns a SQLAlchemy Session() instance.
	"""			"""
	# pylint: disable=global-statement			# pylint: disable=global-statement
	global db			global db

	if db is not None and not reinit:			if db is not None and not reinit:
	return db			return db

	db_uri = conf.get('ldap', 'auth_cache_uri')			db_uri = conf.get('ldap', 'auth_cache_uri')

				if db_uri == "null":
				return None

	if db_uri is None:			if db_uri is None:
	db_uri = 'sqlite:///%s/auth_cache.db' % (KOLAB_LIB_PATH)			db_uri = 'sqlite:///%s/auth_cache.db' % (KOLAB_LIB_PATH)

	if reinit:			if reinit:
	import os
	if os.path.isfile('%s/auth_cache.db' % (KOLAB_LIB_PATH)):			if os.path.isfile('%s/auth_cache.db' % (KOLAB_LIB_PATH)):
	os.unlink('%s/auth_cache.db' % (KOLAB_LIB_PATH))			os.unlink('%s/auth_cache.db' % (KOLAB_LIB_PATH))

	echo = conf.debuglevel > 8			echo = conf.debuglevel > 8
	engine = create_engine(db_uri, echo=echo)			engine = create_engine(db_uri, echo=echo)
	DeclarativeBase.metadata.create_all(engine)			DeclarativeBase.metadata.create_all(engine)

	Session = sessionmaker(bind=engine)			Session = sessionmaker(bind=engine)
	db = Session()			db = Session()
	purge_entries(db)

	return db			return db