4119519683

Build # 4119519683

Build Type

push

github

Committed by GitHub

Commit Message

Lintify (#287)

Run Details

36 of 36 new or added lines in 9 files covered. (100.0%)

5321 of 6332 relevant lines covered (84.03%)

1.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.09

/winterdrp/processors/database/base_database_processor.py

"""
Module containing base database processor class
"""
import logging
from abc import ABC
from pathlib import Path
from typing import Optional

from winterdrp.paths import max_n_cpu
from winterdrp.processors.base_processor import BaseProcessor
from winterdrp.processors.database.postgres import (
    POSTGRES_DUPLICATE_PROTOCOLS,
    DataBaseError,
    PostgresAdmin,
    PostgresUser,
)

logger = logging.getLogger(__name__)


class BaseDatabaseProcessor(BaseProcessor, ABC):
    """Base class for processors which interact with a postgres database"""

    max_n_cpu = min(max_n_cpu, 10)

    def __init__(
        self,
        db_name: str,
        db_table: str,
        schema_path: str | Path,
        pg_user: PostgresUser = PostgresUser(),
        pg_admin: PostgresAdmin = PostgresAdmin(),
        has_foreign_keys: bool = False,
        schema_dir: Optional[str | Path] = None,
        duplicate_protocol: str = "fail",
        q3c_bool: bool = False,
    ):
        super().__init__()
        self.db_name = db_name
        self.db_table = db_table

        self.schema_path = schema_path
        self.schema_dir = Path(schema_dir) if schema_dir is not None else None

        self.pg_user = pg_user
        self._pg_admin = pg_admin

        self.has_foreign_keys = has_foreign_keys
        self.db_check_bool = False
        self.duplicate_protocol = duplicate_protocol
        self.q3c = q3c_bool

        assert self.duplicate_protocol in POSTGRES_DUPLICATE_PROTOCOLS

    def db_exists(self) -> bool:
        """
        Checks if a database exists

        :return: boolean
        """
        return self._pg_admin.check_if_db_exists(db_name=self.db_name)

    def _make_db(self):
        """
        Creates a database

        :return:
        """
        self._pg_admin.create_db(db_name=self.db_name)

    def pg_user_exists(self):
        """
        Check if the specified db user exists

        :return: boolean
        """
        return self._pg_admin.check_if_user_exists(self.pg_user.db_user)

    def _make_pg_user(self):
        """
        Creates a new database user

        :return: None
        """
        self._pg_admin.create_new_user(
            new_db_user=self.pg_user.db_user, new_password=self.pg_user.db_password
        )

    def _grant_privileges_to_pg_user(self):
        """
        Grants db privilege to user

        :return: None
        """
        self._pg_admin.grant_privileges(self.db_name, db_user=self.pg_user.db_user)

    def table_exists(self):
        """
        Check if the database table exists

        :return: boolean
        """
        return self.pg_user.check_if_table_exists(
            db_name=self.db_name,
            db_table=self.db_table,
        )

    def _create_table(self, schema_path: str | Path):
        """
        Makes a database table using schema

        :param schema_path: Path of schema file
        :return: None
        """
        self.pg_user.create_table(
            schema_path,
            db_name=self.db_name,
        )

    def check_prerequisites(
        self,
    ):
        if not self.db_check_bool:
            self.check_database_setup()
            self.db_check_bool = True

    def check_database_setup(self):
        """
        Checks the database is set up, and if not, creates the database and tables

        :return: None
        """
        # If not specified, admin credentials equal user credentials
        # Check that the credentials work at all

        self._pg_admin.validate_credentials()

        # Create pg user for the first time

        if not self.pg_user_exists():
            self._make_pg_user()
        self.pg_user.validate_credentials()

        # Check there is a db

        if not self.db_exists():
            self._make_db()
            self._grant_privileges_to_pg_user()

            if self.has_foreign_keys:
                if self.schema_dir is None:
                    self.schema_dir = self.schema_path.parent
                    logger.warning(
                        f"Warning, integrated db setup with foreign keys requested, "
                        f"but no schema directory specified. "
                        f"Will search for schema files in {self.schema_dir} directory."
                    )

                logger.debug(
                    f"Looking in {self.schema_dir} directory to search for schema files"
                )

                self.pg_user.create_tables_from_schema(self.schema_dir, self.db_name)

                if self.q3c:
                    q3c_dir = self.schema_dir.joinpath("q3c")
                    q3c_indexes_file = q3c_dir.joinpath("q3c_indexes.sql")
                    self.pg_user.run_sql_command_from_file(
                        file_path=q3c_indexes_file, db_name=self.db_name
                    )
                    logger.debug("Created q3c_bool indexes")

        if not self.table_exists():
            self._create_table(self.schema_path)

            if self.q3c:
                q3c_dir = self.schema_dir.joinpath("q3c")
                table_q3c_path = q3c_dir.joinpath(f"q3c_{self.db_table}.sql")

                if not table_q3c_path.exists():
                    err = (
                        f"q3c_bool extension requested but no "
                        f"{table_q3c_path} file found. Please add it in."
                    )
                    logger.error(err)
                    raise DataBaseError(err)

                self.pg_user.run_sql_command_from_file(
                    file_path=table_q3c_path, db_name=self.db_name
                )

1	"""
2	Module containing base database processor class
3	"""
4	import logging	2✔
5	from abc import ABC	2✔
6	from pathlib import Path	2✔
7	from typing import Optional	2✔
8
9	from winterdrp.paths import max_n_cpu	2✔
10	from winterdrp.processors.base_processor import BaseProcessor	2✔
11	from winterdrp.processors.database.postgres import (	2✔
12	POSTGRES_DUPLICATE_PROTOCOLS,
13	DataBaseError,
14	PostgresAdmin,
15	PostgresUser,
16	)
17
18	logger = logging.getLogger(__name__)	2✔
19
20
21	class BaseDatabaseProcessor(BaseProcessor, ABC):	2✔
22	"""Base class for processors which interact with a postgres database"""
23
24	max_n_cpu = min(max_n_cpu, 10)	2✔
25
26	def __init__(	2✔
27	self,
28	db_name: str,
29	db_table: str,
30	schema_path: str \| Path,
31	pg_user: PostgresUser = PostgresUser(),
32	pg_admin: PostgresAdmin = PostgresAdmin(),
33	has_foreign_keys: bool = False,
34	schema_dir: Optional[str \| Path] = None,
35	duplicate_protocol: str = "fail",
36	q3c_bool: bool = False,
37	):
38	super().__init__()	2✔
39	self.db_name = db_name	2✔
40	self.db_table = db_table	2✔
41
42	self.schema_path = schema_path	2✔
43	self.schema_dir = Path(schema_dir) if schema_dir is not None else None	2✔
44
45	self.pg_user = pg_user	2✔
46	self._pg_admin = pg_admin	2✔
47
48	self.has_foreign_keys = has_foreign_keys	2✔
49	self.db_check_bool = False	2✔
50	self.duplicate_protocol = duplicate_protocol	2✔
51	self.q3c = q3c_bool	2✔
52
53	assert self.duplicate_protocol in POSTGRES_DUPLICATE_PROTOCOLS	2✔
54
55	def db_exists(self) -> bool:	2✔
56	"""
57	Checks if a database exists
58
59	:return: boolean
60	"""
61	return self._pg_admin.check_if_db_exists(db_name=self.db_name)	2✔
62
63	def _make_db(self):	2✔
64	"""
65	Creates a database
66
67	:return:
68	"""
69	self._pg_admin.create_db(db_name=self.db_name)	2✔
70
71	def pg_user_exists(self):	2✔
72	"""
73	Check if the specified db user exists
74
75	:return: boolean
76	"""
77	return self._pg_admin.check_if_user_exists(self.pg_user.db_user)	2✔
78
79	def _make_pg_user(self):	2✔
80	"""
81	Creates a new database user
82
83	:return: None
84	"""
85	self._pg_admin.create_new_user(	×
86	new_db_user=self.pg_user.db_user, new_password=self.pg_user.db_password
87	)
88
89	def _grant_privileges_to_pg_user(self):	2✔
90	"""
91	Grants db privilege to user
92
93	:return: None
94	"""
95	self._pg_admin.grant_privileges(self.db_name, db_user=self.pg_user.db_user)	2✔
96
97	def table_exists(self):	2✔
98	"""
99	Check if the database table exists
100
101	:return: boolean
102	"""
103	return self.pg_user.check_if_table_exists(	2✔
104	db_name=self.db_name,
105	db_table=self.db_table,
106	)
107
108	def _create_table(self, schema_path: str \| Path):	2✔
109	"""
110	Makes a database table using schema
111
112	:param schema_path: Path of schema file
113	:return: None
114	"""
115	self.pg_user.create_table(	2✔
116	schema_path,
117	db_name=self.db_name,
118	)
119
120	def check_prerequisites(	2✔
121	self,
122	):
123	if not self.db_check_bool:	2✔
124	self.check_database_setup()	2✔
125	self.db_check_bool = True	2✔
126
127	def check_database_setup(self):	2✔
128	"""
129	Checks the database is set up, and if not, creates the database and tables
130
131	:return: None
132	"""
133	# If not specified, admin credentials equal user credentials
134	# Check that the credentials work at all
135
136	self._pg_admin.validate_credentials()	2✔
137
138	# Create pg user for the first time
139
140	if not self.pg_user_exists():	2✔
141	self._make_pg_user()	×
142	self.pg_user.validate_credentials()	2✔
143
144	# Check there is a db
145
146	if not self.db_exists():	2✔
147	self._make_db()	2✔
148	self._grant_privileges_to_pg_user()	2✔
149
150	if self.has_foreign_keys:	2✔
151	if self.schema_dir is None:	2✔
152	self.schema_dir = self.schema_path.parent	×
153	logger.warning(	×
154	f"Warning, integrated db setup with foreign keys requested, "
155	f"but no schema directory specified. "
156	f"Will search for schema files in {self.schema_dir} directory."
157	)
158
159	logger.debug(	2✔
160	f"Looking in {self.schema_dir} directory to search for schema files"
161	)
162
163	self.pg_user.create_tables_from_schema(self.schema_dir, self.db_name)	2✔
164
165	if self.q3c:	2✔
166	q3c_dir = self.schema_dir.joinpath("q3c")	×
167	q3c_indexes_file = q3c_dir.joinpath("q3c_indexes.sql")	×
168	self.pg_user.run_sql_command_from_file(	×
169	file_path=q3c_indexes_file, db_name=self.db_name
170	)
171	logger.debug("Created q3c_bool indexes")	×
172
173	if not self.table_exists():	2✔
174	self._create_table(self.schema_path)	2✔
175
176	if self.q3c:	2✔
177	q3c_dir = self.schema_dir.joinpath("q3c")	×
178	table_q3c_path = q3c_dir.joinpath(f"q3c_{self.db_table}.sql")	×
179
180	if not table_q3c_path.exists():	×
181	err = (
182	f"q3c_bool extension requested but no "
183	f"{table_q3c_path} file found. Please add it in."
184	)
185	logger.error(err)
186	raise DataBaseError(err)
187
188	self.pg_user.run_sql_command_from_file(	×
189	file_path=table_q3c_path, db_name=self.db_name
190	)

winter-telescope / winterdrp / 4119519683

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous