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!!
