KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use to_schema method in pandera

Best Python code snippet using pandera_python

write-transforms.py

Source:write-transforms.py

1#!/usr/bin/env python32# -*- coding: utf-8 -*-3# Copyright (c) 2018 Bhojpur Consulting Private Limited, India. All rights reserved.4#5# Permission is hereby granted, free of charge, to any person obtaining a copy6# of this software and associated documentation files (the "Software"), to deal7# in the Software without restriction, including without limitation the rights8# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell9# copies of the Software, and to permit persons to whom the Software is10# furnished to do so, subject to the following conditions:11#12# The above copyright notice and this permission notice shall be included in13# all copies or substantial portions of the Software.14#15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN21# THE SOFTWARE.22# Bhojpur ODE Data Model transforms23from collections import defaultdict24from os import listdir25from re import compile26# note the schemas (in reverse order) and the transforms among them27schemas = []28transforms = defaultdict(list)29# report the quality of a single transform30def quality(from_schema, to_schema):31    if from_schema < to_schema:32        return 433    elif to_schema < '2008':34        return 135    elif to_schema < '2010-06':36        return 237    else:38        return 339# find the shortest sequence of transforms of sufficient quality40def shortest_path(from_schema, to_schema, min_quality):41    paths = [[from_schema]]42    while True:43        # any paths lead to the goal?44        for path in paths:45            if path[-1] == to_schema:  # yes, goal path found46                return path47        # extend all the current paths48        next_paths = []49        for path in paths:50            curr_step = path[-1]51            for next_step in transforms[curr_step]:52                if quality(curr_step, next_step) >= min_quality and \53                        next_step not in path:54                    next_path = list(path)55                    next_path.append(next_step)56                    next_paths.append(next_path)57        if not next_paths:  # no more transforms to try58            return None59        paths = next_paths60# the style of transform file names61name_pattern = compile('^(.+)\-to\-(.+)\.xsl$')62# scan the current directory to determine the schemas and transforms63def load_transforms():64    global transforms65    global schemas66    seen = set()67    for name in listdir('.'):68        match = name_pattern.match(name)69        if match:70            from_schema = match.group(1)71            to_schema = match.group(2)72            transforms[from_schema].append(to_schema)73            seen.add(from_schema)74            seen.add(to_schema)75    schemas = list(seen)76    schemas.sort(reverse=True)77# note of the best sequence of transforms among the schemas78best_paths = {}79# determine the best sequences of transforms among the schemas80load_transforms()81for from_schema in schemas:82    for to_schema in schemas:83        if from_schema != to_schema:84            for min_quality in range(4, 0, -1):85                path = shortest_path(from_schema, to_schema, min_quality)86                if path:87                    best_paths[(from_schema, to_schema)] = (path, min_quality)88                    break89            if not best_paths[(from_schema, to_schema)]:90                raise Exception(91                    'no path from ' + from_schema + ' to ' + to_schema)92# name the quality levels of transforms93qualities = ['poor', 'fair', 'good', 'excellent']94# print in XML a sequence of transforms to the given schema95def print_path(to_schema):96    (path, min_quality) = best_paths[(from_schema, to_schema)]97    print ('\t\t\t<target schema="' + to_schema + \98        '" quality="' + qualities[min_quality - 1] + '">')99    while len(path) > 1:100        print ('\t\t\t\t<transform file="' + \101            path[0] + '-to-' + path[1] + '.xsl"/>')102        path = path[1:]103    print('\t\t\t</target>')104# print in XML all the transforms among the schemas105print("""<?xml version = "1.0" encoding = "UTF-8"?>106<!--107  #%L108  Bhojpur ODE Data Model transforms109  %%110 Copyright (c) 2018 Bhojpur Consulting Private Limited, India. All rights reserved.111  %%112 Permission is hereby granted, free of charge, to any person obtaining a copy113 of this software and associated documentation files (the "Software"), to deal114 in the Software without restriction, including without limitation the rights115 to use, copy, modify, merge, publish, distribute, sublicense, and/or sell116 copies of the Software, and to permit persons to whom the Software is117 furnished to do so, subject to the following conditions:118 The above copyright notice and this permission notice shall be included in119 all copies or substantial portions of the Software.120 THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR121 IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,122 FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE123 AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER124 LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,125 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN126 THE SOFTWARE.127  -->128""")129print('<ode-transforms current="' + schemas[0] + '">')130for from_schema in schemas:131    print('\t<source schema="' + from_schema + '">')132    print('\t\t<upgrades>')133    upgrades = True134    for to_schema in schemas:135        if from_schema < to_schema:136            print_path(to_schema)137        elif to_schema < from_schema:138            if upgrades:139                upgrades = False140                print('\t\t</upgrades>')141                print('\t\t<downgrades>')142            print_path(to_schema)143    if upgrades:144        print('\t\t</upgrades>')145        print('\t\t<downgrades>')146    print('\t\t</downgrades>')147    print('\t</source>')...

sql_ddl.py

Source:sql_ddl.py

1from __future__ import annotations2from sqlalchemy.ext.compiler import compiles3from sqlalchemy.schema import DDLElement4from sqlalchemy.sql import Select5__all__ = [6    "CreateSchema",7    "DropSchema",8    "RenameSchema",9    "CreateTableAsSelect",10    "CopyTable",11    "DropTable",12]13class CreateSchema(DDLElement):14    def __init__(self, name, if_not_exists=False):15        self.name = name16        self.if_not_exists = if_not_exists17class DropSchema(DDLElement):18    def __init__(self, name, if_exists=False, cascade=False):19        self.name = name20        self.if_exists = if_exists21        self.cascade = cascade22class RenameSchema(DDLElement):23    def __init__(self, _from, to):24        self._from = _from25        self.to = to26class CreateTableAsSelect(DDLElement):27    def __init__(self, name: str, schema: str, query: Select):28        self.name = name29        self.schema = schema30        self.query = query31class CopyTable(DDLElement):32    def __init__(self, from_name, from_schema, to_name, to_schema, if_not_exists=False):33        self.from_name = from_name34        self.from_schema = from_schema35        self.to_name = to_name36        self.to_schema = to_schema37        self.if_not_exists = if_not_exists38class DropTable(DDLElement):39    def __init__(self, name, schema, if_exists=False):40        self.name = name41        self.schema = schema42        self.if_exists = if_exists43@compiles(CreateSchema)44def visit_create_schema(create: CreateSchema, compiler, **kw):45    schema = compiler.preparer.format_schema(create.name)46    text = ["CREATE SCHEMA"]47    if create.if_not_exists:48        text.append("IF NOT EXISTS")49    text.append(schema)50    return " ".join(text)51@compiles(DropSchema)52def visit_drop_schema(drop: DropSchema, compiler, **kw):53    schema = compiler.preparer.format_schema(drop.name)54    text = ["DROP SCHEMA"]55    if drop.if_exists:56        text.append("IF EXISTS")57    text.append(schema)58    if drop.cascade:59        text.append("CASCADE")60    return " ".join(text)61@compiles(RenameSchema)62def visit_rename_schema(rename: RenameSchema, compiler, **kw):63    _from = compiler.preparer.format_schema(rename._from)64    to = compiler.preparer.format_schema(rename.to)65    return "ALTER SCHEMA " + _from + " RENAME TO " + to66@compiles(CreateTableAsSelect)67def visit_create_table_as_select(create: CreateTableAsSelect, compiler, **kw):68    name = compiler.preparer.quote_identifier(create.name)69    schema = compiler.preparer.format_schema(create.schema)70    kw["literal_binds"] = True71    select = compiler.sql_compiler.process(create.query, **kw)72    return f"CREATE TABLE {schema}.{name} AS\n{select}"73@compiles(CopyTable)74def visit_copy_table(copy: CopyTable, compiler, *kw):75    from_name = compiler.preparer.quote_identifier(copy.from_name)76    to_name = compiler.preparer.quote_identifier(copy.to_name)77    from_schema = compiler.preparer.format_schema(copy.from_schema)78    to_schema = compiler.preparer.format_schema(copy.to_schema)79    text = ["CREATE TABLE"]80    if copy.if_not_exists:81        text.append("IF NOT EXISTS")82    text.append(f"{to_schema}.{to_name}")83    text.append("AS")84    return " ".join(text) + f"\nSELECT * FROM {from_schema}.{from_name}"85@compiles(DropTable)86def visit_drop_table(drop: DropTable, compiler, **kw):87    table = compiler.preparer.quote_identifier(drop.name)88    schema = compiler.preparer.format_schema(drop.schema)89    text = ["DROP TABLE"]90    if drop.if_exists:91        text.append("IF EXISTS")92    text.append(f"{schema}.{table}")...

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.