Best Python code snippet using tempest_python
Schema.py
Source:Schema.py
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:...
executor.py
Source:executor.py
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)...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!