Source code for pystratum_mssql.MsSqlMetadataDataLayer

from typing import Any, Dict, List

from pystratum_backend.StratumStyle import StratumStyle
from pystratum_common.MetadataDataLayer import MetadataDataLayer

from pystratum_mssql.MsSqlDataLayer import MsSqlDataLayer
from pystratum_mssql.MsSqlConnector import MsSqlConnector


[docs]class MsSqlMetadataDataLayer(MetadataDataLayer): """ Data layer for retrieving metadata and loading stored routines. """ __dl = None """ The connection to the SQL Server instance. :type: pystratum_mssql.MsSqlDataLayer.MsSqlDataLayer|None """ # ------------------------------------------------------------------------------------------------------------------ def __init__(self, io: StratumStyle, connector: MsSqlConnector): """ Object constructor. :param PyStratumStyle io: The output decorator. """ super().__init__(io) self.__dl: MsSqlDataLayer = MsSqlDataLayer(connector) """ The connection to the MySQL instance. """ # ------------------------------------------------------------------------------------------------------------------
[docs] def commit(self) -> None: """ Connects to a SQL Server instance. """ self.__dl.commit()
# ------------------------------------------------------------------------------------------------------------------
[docs] def connect(self) -> None: """ Connects to a SQL Server instance. """ self.__dl.connect()
# ------------------------------------------------------------------------------------------------------------------
[docs] def disconnect(self) -> None: """ Disconnects from the SQL Server instance. """ self.__dl.disconnect()
# ------------------------------------------------------------------------------------------------------------------
[docs] def drop_stored_routine(self, routine_type: str, schema_name: str, routine_name: str) -> None: """ Drops a stored routine if it exists. :param str routine_type: The type of the routine (i.e. procedure or function). :param str schema_name: The name of the schema. :param str routine_name: The name of the routine. """ sql = "drop {0} [{1}].[{2}]".format(routine_type, schema_name, routine_name) self.execute_none(sql)
# ------------------------------------------------------------------------------------------------------------------
[docs] def drop_temporary_table(self, table_name: str) -> None: """ Drops a temporary table. :param str table_name: The name of the table. """ sql = 'drop temporary table `{0}`'.format(table_name) self.execute_none(sql)
# ------------------------------------------------------------------------------------------------------------------
[docs] def execute_none(self, query: str) -> None: """ Executes a query that does not select any rows. :param str query: The query. :rtype: int """ self._log_query(query) return self.__dl.execute_none(query)
# ------------------------------------------------------------------------------------------------------------------
[docs] def execute_rows(self, query: str) -> List[Dict[str, Any]]: """ Executes a query that selects 0 or more rows. Returns the selected rows (an empty list if no rows are selected). :param str query: The query. :rtype: list[dict[str,*]] """ self._log_query(query) return self.__dl.execute_rows(query)
# ------------------------------------------------------------------------------------------------------------------
[docs] def get_all_table_columns(self) -> List[Dict[str, Any]]: """ Selects metadata of all columns of all tables. :rtype: list[dict[str,*]] """ sql = """ select scm.name as schema_name , tab.name as table_name , col.name as column_name , isnull(stp.name,utp.name) as data_type , col.max_length , col.precision , col.scale , col.column_id from sys.columns col inner join sys.types utp on utp.user_type_id = col.user_type_id and utp.system_type_id = col.system_type_id left outer join sys.types stp on utp.is_user_defined = 1 and stp.is_user_defined = 0 and utp.system_type_id = stp.system_type_id and utp.user_type_id <> stp.user_type_id and stp.user_type_id = stp.system_type_id inner join sys.tables tab on col.object_id = tab.object_id inner join sys.schemas scm on tab.schema_id = scm.schema_id where tab.type in ('U','S','V') order by scm.name , tab.name , col.column_id """ return self.execute_rows(sql)
# ------------------------------------------------------------------------------------------------------------------
[docs] def get_label_tables(self, regex: str) -> List[Dict[str, Any]]: """ Selects metadata of tables with a label column. :param str regex: The regular expression for columns which we want to use. :rtype: list[dict[str,*]] """ sql = """ select db_name() [database] , scm.name [schema_name] , tab.name [table_name] , cl1.name [label] , cl2.name [id] from sys.schemas scm inner join sys.tables tab on tab.[schema_id] = scm.[schema_id] inner join sys.all_columns cl1 on cl1.[object_id] = tab.[object_id] inner join sys.all_columns cl2 on cl2.[object_id] = tab.[object_id] where cl1.name like '{0}' and cl2.is_identity = 1""".format(regex) return self.execute_rows(sql)
# ------------------------------------------------------------------------------------------------------------------
[docs] def get_labels_from_table(self, database_name: str, schema_name: str, table_name: str, id_column_name: str, label_column_name: str) -> List[Dict[str, Any]]: """ Selects all labels from a table with labels. :param str database_name: The name of the database. :param str schema_name: The name of the schema. :param str table_name: The name of the table. :param str id_column_name: The name of the auto increment column. :param str label_column_name: The name of the column with labels. :rtype: list[dict[str,*]] """ sql = """ select tab.[{0!s}] id , tab.[{1!s}] label from [{2!s}].[{3!s}].[{4!s}] tab where nullif(tab.[{1!s}],'') is not null""".format(id_column_name, label_column_name, database_name, schema_name, table_name) return self.execute_rows(sql)
# ------------------------------------------------------------------------------------------------------------------
[docs] def get_routine_parameters(self, schema_name: str, routine_name: str) -> List[Dict[str, Any]]: """ Selects metadata of the parameters of a stored routine. :param str schema_name: The name of the schema. :param str routine_name: The name of the routine. :rtype: list[dict[str,*]] """ sql = """ select par.name parameter_name , typ.name type_name , typ.max_length , typ.precision , typ.scale from sys.schemas scm inner join sys.all_objects prc on prc.[schema_id] = scm.[schema_id] inner join sys.all_parameters par on par.[object_id] = prc.[object_id] inner join sys.types typ on typ.user_type_id = par.system_type_id where scm.name = '%s' and prc.name = '%s' order by par.parameter_id""" % (schema_name, routine_name) return self.execute_rows(sql)
# ------------------------------------------------------------------------------------------------------------------
[docs] def get_routines(self) -> List[Dict[str, Any]]: """ Selects metadata of all routines. :rtype: list[dict[str,*]] """ sql = """ select scm.name as schema_name , prc.name as routine_name , prc.type as routine_type from sys.all_objects prc inner join sys.schemas scm on scm.schema_id = prc.schema_id where prc.type in ('P','FN','TF') and scm.name <> 'sys' and prc.is_ms_shipped=0""" return self.execute_rows(sql)
# ----------------------------------------------------------------------------------------------------------------------