How to use _table_columns method in tempest

Best Python code snippet using tempest_python

Schema.py

Source:Schema.py Github

copy

Full Screen

1#!/usr/bin/env python2# -*- coding: utf-8 -*-3#4# Copyright [2017] Tatarnikov Viktor [viktor@tatarnikov.org]5#6# Licensed under the Apache License, Version 2.0 (the "License");7# you may not use this file except in compliance with the License.8# You may obtain a copy of the License at9#10# http://www.apache.org/licenses/LICENSE-2.011#12# Unless required by applicable law or agreed to in writing, software13# distributed under the License is distributed on an "AS IS" BASIS,14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.15# See the License for the specific language governing permissions and16# limitations under the License.17import sys18reload(sys)19sys.setdefaultencoding('utf8')20from collections import OrderedDict21class Schema:22 """23 Класс хелпер, анализирует содержимое класса, если это описание таблицы ORM SqlAlchemy, то скрывает из Sphinx'a AutoDoc'a столбцы и связи, а потом выводит их в удобной таблице в шапке класса.24 А также выводит SQL код создания таблицы.25 26 .. todo: Добавить графическую отрисовку связей таблиц.27 28 """29 def __init__(self, object):30 self.object = object31 self._table_name = "undef"32 self._table_args = {}33 self._table_engine = "undef"34 self._table_description = "undef"35 self._table_charset = "undef"36 self._table_collate = "undef"37 self._table_module = ""38 self._table_class = ""39 40 self._table_columns = {}41 self._table_links = {}42 43 def rst_name(self, name):44 """45 Функция переименования названия модуля под ссылку в rst46 47 :param name: Имя модуля или класса48 :type name: String49 :return: Имя ссылки в rst50 :rtype: String51 """52 return name.lower().replace(".", "-")53 def has_schema(self):54 """55 Функция определения, является ли обьект схемой ORM SqlAlchemy.56 57 :return: True если это класс схемы ORM SqlAlchemy, иначе False58 :rtype: Boolean59 """60 attrs = ["__tablename__", "__table_args__"]61 62 for attr in attrs:63 if not hasattr(self.object, attr):64 return False65 66 return True67 def has_schema_attr(self):68 """69 Функция определения, является ли обьект атрибутом класса ORM SqlAlchemy.70 71 :return: True если это атрибут схемы ORM SqlAlchemy, иначе False72 :rtype: Boolean73 """74 if hasattr(self.object, 'prop'):75 if hasattr(self.object.prop, 'strategy_wildcard_key'):76 if self.object.prop.strategy_wildcard_key in ['column', 'relationship']:77 return True78 79 return False80 def get_attr_order(self, key):81 """82 Функция определения порядкового номера в сортировке столбцов таблицы по имени столбца.83 84 :param key: Название столбца85 :type key: String86 :return: Порядковый номер сортировки87 :rtype: Integer88 """89 key_prefix = key.split("_")[0]90 91 if key_prefix in ['id']:92 return 093 elif key_prefix in ['cd']:94 return 195 elif key_prefix in ['is']:96 return 297 elif key_prefix in ['web']:98 return 499 elif key_prefix in ['date']:100 return 5101 else:102 return 3103 def parse_table(self):104 """105 Функция анализа класса и получения данных о схеме.106 107 :return: None108 :rtype: Nothing109 """110 self._table_name = getattr(self.object, "__tablename__", self._table_name)111 self._table_args = getattr(self.object, "__table_args__", self._table_args)112 self._table_module = getattr(self.object, "__module__", self._table_module)113 self._table_class = getattr(self.object, "__name__", self._table_class)114 115 if "mysql_engine" in self._table_args:116 self._table_engine = self._table_args['mysql_engine']117 118 if "mysql_charset" in self._table_args:119 self._table_charset = self._table_args['mysql_charset']120 121 if "mysql_collate" in self._table_args:122 self._table_collate = self._table_args['mysql_collate']123 124 if "mysql_comment" in self._table_args:125 self._table_description = self._table_args['mysql_comment']126 def parse_columns(self):127 """128 Функция анализа класса и получения данных о схеме.129 130 :return: None131 :rtype: Nothing132 """133 for attr_key in self.object.__dict__.keys():134 if attr_key in self.object._sa_class_manager:135 136 if self.object._sa_class_manager[attr_key].prop.strategy_wildcard_key == 'column':137 self._table_columns[attr_key] = {}138 self._table_columns[attr_key]['order'] = self.get_attr_order(attr_key)139 self._table_columns[attr_key]['name'] = attr_key140 self._table_columns[attr_key]['description'] = str(self.object._sa_class_manager[attr_key].prop.doc).replace(",", "")141 self._table_columns[attr_key]['type'] = str(self.object._sa_class_manager[attr_key].prop.columns[0].type).lower()142 if self.object._sa_class_manager[attr_key].prop.strategy_wildcard_key == 'relationship':143 # Мои зависимости сугубо с маленькой буквы144 if 97 <= ord(attr_key[0]) <= 122:145 self._table_links[attr_key] = {}146 self._table_links[attr_key]['name'] = attr_key147 self._table_links[attr_key]['description'] = str(self.object._sa_class_manager[attr_key].prop.doc).replace(",", "")148 self._table_links[attr_key]['type'] = "relationship"149 self._table_links[attr_key]['table'] = self.object._sa_class_manager[attr_key].prop.mapper.entity.__name__150 self._table_links[attr_key]['module'] = self.object._sa_class_manager[attr_key].prop.mapper.entity.__module__151 152 def render(self, lines):153 """154 Функция отрисовки данных в rst. Функция ничего не возвращает, а заменяет данные в массиве.155 156 :param lines: Массив строк с описанием класса от AutoDoc'a Sphinx'a157 :type lines: Array of string158 :return: None159 :rtype: Nothing160 """161 if self.has_schema():162 lines_original = []163 164 # Зануляем содержиное165 for line_index in xrange(len(lines)):166 lines_original.append(str(lines[line_index]))167 lines[line_index] = ""168 169 #170 # Отрисовка шапки171 #172 lines.append(u"""Класс '``{tclass}``' определяет структуру SQL таблицы '``{tname}``' [Engine: {tengine}; Charset: {tcharset}; Collate: {tcollate}]. """.format(tclass=self._table_class, tname=self._table_name, tengine=self._table_engine, tcharset=self._table_charset, tcollate=self._table_collate))173 lines.append(u"""""")174 175 lines.append(u"""Класс содержит: столбцов ``{}``; подключенных связей ``{}``.""".format(len(self._table_columns), len(self._table_links)))176 lines.append(u"""""")177 178 #179 # Отрисовка таблицы180 #181 lines.append(u""".. rubric:: Логическое описание полей таблицы.""")182 lines.append(u"""""")183 lines.append(u""".. _schema-{}-{}:""".format(self.rst_name(self._table_module), self.rst_name(self._table_class)))184 lines.append(u""".. cssclass:: table-striped""")185 186 lines.append(u""".. csv-table:: SQL Table: {} ({})""".format(self._table_class, self._table_description))187 lines.append(u""" :header: Название, Тип, Описание""")188 lines.append(u""" :widths: 10, 10, 50""")189 lines.append(u""" :stub-columns: 1""")190 lines.append(u"""""")191 192 columns_keys = OrderedDict(sorted(self._table_columns.items(), key=lambda t: (t[1]['order'], t[0])))193 194 for column_key in columns_keys:195 column = self._table_columns[column_key]196 197 if column['description'] == "":198 column['description'] = " "199 lines.append(u""" {}, {}, {}""".format(column['name'], column['type'], column['description']))200 201 links_keys = OrderedDict(sorted(self._table_links.items(), key=lambda t: (t[1]['name'], t[0])))202 203 for link_key in links_keys:204 link = self._table_links[link_key]205 206 lines.append(u""" {}, {}, :ref:`schema-{}-{}`""".format(link['name'], link['type'], self.rst_name(link['module']), self.rst_name(link['table'])))207 lines.append(u"""""")208 209 #210 # Генерация SQL211 #212 from sqlalchemy.schema import CreateTable213 214 sql_raw = str(CreateTable(self.object.__table__))215 lines.append(u""".. rubric:: Vanila SQL: Создание таблицы ``{}``.""".format(self._table_name))216 lines.append(u"""""")217 lines.append(u""".. code-block:: sql """)218 lines.append(u""" :linenos:""")219 220 for line in sql_raw.splitlines():221 lines.append(u""" {}""".format(line))222 223 lines.append(u"""""")224 225 #226 # Описание из автодокументации227 #228 lines.append(u""".. rubric:: Описание класса ``{}``.""".format(self._table_class))229 lines.append(u"""""")230 for line_index in xrange(len(lines_original)):231 lines.append(lines_original[line_index])232 lines.append(u"""""")233 234 #235 # Добавление рубрики236 #237 lines.append(u""".. rubric:: Бизнес логика схемы ``{}``.""".format(self._table_class))238 lines.append(u"""""")239if __name__ == "__main__":240 from Interfaces.MySQL import Schema as sqlSchemna241 lines = []242 schema = Schema(object=sqlSchemna.WebSite)243 if schema.has_schema():244 print "Object is Schema."245 schema.parse_table()246 schema.parse_columns()247 schema.render(lines=lines)248 print ": reStructuredText Start"249 print u"\n".join(lines)250 print ": reStructuredText End"251 252 elif schema.has_schema_attr():253 print "Object is Schema Attribute."254 255 else:...

Full Screen

Full Screen

executor.py

Source:executor.py Github

copy

Full Screen

1# (C) 2022 GoodData Corporation2from __future__ import annotations3from typing import Any, Generator, NamedTuple, Optional4import gooddata_fdw.column_utils as column_utils5import gooddata_fdw.column_validation as col_val6from gooddata_fdw.environment import ColumnDefinition, Qual7from gooddata_fdw.filter import extract_filters_from_quals8from gooddata_fdw.options import ServerOptions, TableOptions9from gooddata_fdw.result_reader import InsightTableResultReader, TableResultReader10from gooddata_sdk import GoodDataSdk11class InitData(NamedTuple):12 sdk: GoodDataSdk13 server_options: ServerOptions14 table_options: TableOptions15 columns: dict[str, ColumnDefinition]16class Executor:17 def __init__(self, inputs: InitData, column_validators: list[col_val.ColumnValidator]) -> None:18 self._sdk = inputs.sdk19 self._table_columns = inputs.columns20 self._column_validators = column_validators21 @classmethod22 def can_react(cls, inputs: InitData) -> bool:23 return False24 def validate_columns_def(self) -> None:25 for column_name, column_def in self._table_columns.items():26 for validator in self._column_validators:27 validator.validate(column_name, column_def)28 def execute(29 self, quals: list[Qual], columns: list[str], sort_keys: Optional[list[Any]] = None30 ) -> Generator[dict[str, Any], None, None]:31 raise NotImplementedError()32class InsightExecutor(Executor):33 _COLUMN_VALIDATORS = [col_val.LocalIdOptionValidator(), col_val.IdOptionValidator(mandatory=False)]34 def __init__(self, inputs: InitData) -> None:35 super().__init__(inputs, self._COLUMN_VALIDATORS)36 self._workspace = inputs.table_options.workspace37 assert inputs.table_options.insight is not None38 self._insight = inputs.table_options.insight39 self._table_columns = inputs.columns40 @classmethod41 def can_react(cls, inputs: InitData) -> bool:42 return inputs.table_options.insight is not None43 def execute(44 self, quals: list[Qual], columns: list[str], sort_keys: Optional[list[Any]] = None45 ) -> Generator[dict[str, Any], None, None]:46 results_reader = InsightTableResultReader(self._table_columns, columns)47 insight = self._sdk.insights.get_insight(self._workspace, self._insight)48 table = self._sdk.tables.for_insight(self._workspace, insight)49 return results_reader.read_all_rows(table)50class ComputeExecutor(Executor):51 _COLUMN_VALIDATORS: list[col_val.ColumnValidator] = [col_val.IdOptionValidator(mandatory=True)]52 def __init__(self, inputs: InitData) -> None:53 super().__init__(inputs, self._COLUMN_VALIDATORS)54 self._workspace = inputs.table_options.workspace55 self._results_reader = TableResultReader(self._table_columns)56 @classmethod57 def can_react(cls, inputs: InitData) -> bool:58 return inputs.table_options.compute is not None59 def execute(60 self, quals: list[Qual], columns: list[str], sort_keys: Optional[list[Any]] = None61 ) -> Generator[dict[str, Any], None, None]:62 col_val.validate_columns_in_table_def(self._table_columns, columns)63 items = [column_utils.table_col_as_computable(self._table_columns[col_name]) for col_name in columns]64 # TODO: push down more filters that are included in quals65 filters = extract_filters_from_quals(quals, self._table_columns)66 table = self._sdk.tables.for_items(self._workspace, items, filters)67 return self._results_reader.read_all_rows(table)68class CustomExecutor(Executor):69 _COLUMN_VALIDATORS: list[col_val.ColumnValidator] = [col_val.IdOptionValidator(mandatory=True)]70 def __init__(self, inputs: InitData) -> None:71 super().__init__(inputs, self._COLUMN_VALIDATORS)72 self._workspace = inputs.table_options.workspace73 self._results_reader = TableResultReader(self._table_columns)74 @classmethod75 def can_react(cls, inputs: InitData) -> bool:76 return True77 def execute(78 self, quals: list[Qual], columns: list[str], sort_keys: Optional[list[Any]] = None79 ) -> Generator[dict[str, Any], None, None]:80 items = [column_utils.table_col_as_computable(col) for col in self._table_columns.values()]81 # TODO: pushdown more filters that are included in quals82 filters = extract_filters_from_quals(quals, self._table_columns)83 table = self._sdk.tables.for_items(self._workspace, items, filters)84 return self._results_reader.read_all_rows(table)85class ExecutorFactory:86 # Order is important - first executor supporting InitData is used87 _SUPPOERTED_EXECUTORS = [InsightExecutor, ComputeExecutor, CustomExecutor]88 @classmethod89 def create(cls, inputs: InitData) -> Executor:90 for executor in cls._SUPPOERTED_EXECUTORS:91 if executor.can_react(inputs):92 return executor(inputs)...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run tempest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful