Differential D2232 Diff 7718 swh/lister/debian/__init__.py

Changeset View

Standalone View

swh/lister/debian/init.py

	# Copyright (C) 2019 The Software Heritage developers			# Copyright (C) 2019 The Software Heritage developers
	# See the AUTHORS file at the top-level directory of this distribution			# See the AUTHORS file at the top-level directory of this distribution
	# License: GNU General Public License version 3, or any later version			# License: GNU General Public License version 3, or any later version
	# See top-level LICENSE file for more information			# See top-level LICENSE file for more information

	from typing import Any, List, Mapping			from typing import Any, List, Mapping


	def debian_init(db_engine, lister=None,			def debian_init(db_engine,
	override_conf: Mapping[str, Any] = {},			override_conf: Mapping[str, Any] = {},
	distributions: List[str] = ['stretch', 'buster'],			distributions: List[str] = ['stretch', 'buster'],
	area_names: List[str] = ['main', 'contrib', 'non-free']):			area_names: List[str] = ['main', 'contrib', 'non-free']):
	"""Initialize the debian data model.			"""Initialize the debian data model.

	Args:			Args:
	db_engine: SQLAlchemy manipulation database object			db_engine: SQLAlchemy manipulation database object
	lister: Debian lister instance. None by default.
	override_conf: Override conf to pass to instantiate a lister			override_conf: Override conf to pass to instantiate a lister
	distributions: Default distribution to build			distributions: Default distribution to build


	"""			"""
	distribution_name = 'Debian'			distribution_name = 'Debian'
	from swh.lister.debian.models import Distribution, Area			from swh.lister.debian.models import Distribution, Area
				from sqlalchemy.orm import sessionmaker
				db_session = sessionmaker(bind=db_engine)()

	if lister is None:			existing_distrib = db_session \
	from .lister import DebianLister
	lister = DebianLister(distribution=distribution_name,
	override_config=override_conf)

	if not lister.db_session\
	.query(Distribution)\			.query(Distribution) \
	.filter(Distribution.name == distribution_name)\			.filter(Distribution.name == distribution_name) \
	.one_or_none():			.one_or_none()
				if not existing_distrib:
	d = Distribution(			distrib = Distribution(name=distribution_name,
	name=distribution_name,
	type='deb',			type='deb',
	mirror_uri='http://deb.debian.org/debian/')			mirror_uri='http://deb.debian.org/debian/')
	lister.db_session.add(d)			db_session.add(distrib)

	areas = []
	for distribution_name in distributions:			for distribution_name in distributions:
	for area_name in area_names:			for area_name in area_names:
	areas.append(Area(			area = Area(
	name='%s/%s' % (distribution_name, area_name),			name='%s/%s' % (distribution_name, area_name),
	distribution=d,			distribution=distrib,
	))			)
	lister.db_session.add_all(areas)			db_session.add(area)
	lister.db_session.commit()
				db_session.commit()
				db_session.close()


	def register() -> Mapping[str, Any]:			def register() -> Mapping[str, Any]:
	from .lister import DebianLister			from .lister import DebianLister
	return {'models': [DebianLister.MODEL],			return {'models': [DebianLister.MODEL],
	'lister': DebianLister,			'lister': DebianLister,
	'task_modules': ['%s.tasks' % __name__],			'task_modules': ['%s.tasks' % __name__],
	'init': debian_init}			'init': debian_init}