Utility for managing database tables

jdbc_test_tool.py

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

# Part of the Demo Codes for Apache Beam cross-language JDBCIO

import sqlalchemy

def to_url(username, password, endpoint):
  if len(password) > 0:
    return 'postgresql://'+username+':'+password+'@'+endpoint+'/postgres'
  else:
    return 'postgresql://'+endpoint+'/postgres'

def create_table(table_name, schema, username='postgres', password='', endpoint='localhost:5432'):
  engine = sqlalchemy.create_engine(to_url(username, password, endpoint))

  engine.execute(
      "CREATE TABLE IF NOT EXISTS {}{}".format(table_name, schema))

def query(statement,username='postgres', password='', endpoint='localhost:5432'):
  engine = sqlalchemy.create_engine(to_url(username, password, endpoint))

  engine.execute(statement)

def teardown_table(table_name, username='postgres', password='', endpoint='localhost:5432'):
  engine = sqlalchemy.create_engine(to_url(username, password, endpoint))

  engine.execute('DROP TABLE IF EXISTS ' + table_name)