Best Python code snippet using tempest_python
copy_from.py
Source:copy_from.py
1# copy_from.py -- example about copy_from 2#3# Copyright (C) 2002 Tom Jenkins <tjenkins@devis.com>4# Copyright (C) 2005 Federico Di Gregorio <fog@initd.org>5#6# This program is free software; you can redistribute it and/or modify7# it under the terms of the GNU General Public License as published by the8# Free Software Foundation; either version 2, or (at your option) any later9# version.10#11# This program is distributed in the hope that it will be useful, but12# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTIBILITY13# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License14# for more details.15#16## put in DSN your DSN string17DSN = 'dbname=test'18## don't modify anything below tis line (except for experimenting)19import sys20import os21import StringIO22import psycoev23if len(sys.argv) > 1:24 DSN = sys.argv[1]25print "Opening connection using dns:", DSN26conn = psycoev.connect(DSN)27print "Encoding for this connection is", conn.encoding28curs = conn.cursor()29try:30 curs.execute("CREATE TABLE test_copy (fld1 text, fld2 text, fld3 int4)")31except:32 conn.rollback()33 curs.execute("DROP TABLE test_copy")34 curs.execute("CREATE TABLE test_copy (fld1 text, fld2 text, fld3 int4)")35conn.commit()36# copy_from with default arguments, from open file37io = open('copy_from.txt', 'wr')38data = ['Tom\tJenkins\t37\n',39 'Madonna\t\\N\t45\n',40 'Federico\tDi Gregorio\t\\N\n']41io.writelines(data)42io.close()43io = open('copy_from.txt', 'r')44curs.copy_from(io, 'test_copy')45print "1) Copy %d records from file object " % len(data) + \46 "using defaults (sep: \\t and null = \\N)"47io.close()48curs.execute("SELECT * FROM test_copy")49rows = curs.fetchall()50print " Select returned %d rows" % len(rows)51for r in rows:52 print " %s %s\t%s" % (r[0], r[1], r[2])53curs.execute("delete from test_copy")54conn.commit()55# copy_from using custom separator, from open file56io = open('copy_from.txt', 'wr')57data = ['Tom:Jenkins:37\n',58 'Madonna:\N:45\n',59 'Federico:Di Gregorio:\N\n']60io.writelines(data)61io.close()62io = open('copy_from.txt', 'r')63curs.copy_from(io, 'test_copy', ':')64print "2) Copy %d records from file object using sep = :" % len(data)65io.close()66curs.execute("SELECT * FROM test_copy")67rows = curs.fetchall()68print " Select returned %d rows" % len(rows)69for r in rows:70 print " %s %s\t%s" % (r[0], r[1], r[2])71curs.execute("delete from test_copy")72conn.commit()73# copy_from using custom null identifier, from open file74io = open('copy_from.txt', 'wr')75data = ['Tom\tJenkins\t37\n',76 'Madonna\tNULL\t45\n',77 'Federico\tDi Gregorio\tNULL\n']78io.writelines(data)79io.close()80io = open('copy_from.txt', 'r')81curs.copy_from(io, 'test_copy', null='NULL')82print "3) Copy %d records from file object using null = NULL" % len(data)83io.close()84curs.execute("SELECT * FROM test_copy")85rows = curs.fetchall()86print " Select using cursor returned %d rows" % len(rows)87for r in rows:88 print " %s %s\t%s" % (r[0], r[1], r[2])89curs.execute("delete from test_copy")90conn.commit()91# copy_from using custom separator and null identifier92io = open('copy_from.txt', 'wr')93data = ['Tom:Jenkins:37\n', 'Madonna:NULL:45\n', 'Federico:Di Gregorio:NULL\n']94io.writelines(data)95io.close()96io = open('copy_from.txt', 'r')97curs.copy_from(io, 'test_copy', ':', 'NULL')98print "4) Copy %d records from file object " % len(data) + \99 "using sep = : and null = NULL"100io.close()101curs.execute("SELECT * FROM test_copy")102rows = curs.fetchall()103print " Select using cursor returned %d rows" % len(rows)104for r in rows:105 print " %s %s\t%s" % (r[0], r[1], r[2])106curs.execute("delete from test_copy")107conn.commit()108# anything can be used as a file if it has .read() and .readline() methods109data = StringIO.StringIO()110data.write('\n'.join(['Tom\tJenkins\t37',111 'Madonna\t\N\t45',112 'Federico\tDi Gregorio\t\N']))113data.seek(0)114curs.copy_from(data, 'test_copy')115print "5) Copy 3 records from StringIO object using defaults"116curs.execute("SELECT * FROM test_copy")117rows = curs.fetchall()118print " Select using cursor returned %d rows" % len(rows)119for r in rows:120 print " %s %s\t%s" % (r[0], r[1], r[2])121curs.execute("delete from test_copy")122conn.commit()123# simple error test124print "6) About to raise an error"125data = StringIO.StringIO()126data.write('\n'.join(['Tom\tJenkins\t37',127 'Madonna\t\N\t45',128 'Federico\tDi Gregorio\taaa']))129data.seek(0)130try:131 curs.copy_from(data, 'test_copy')132except StandardError, err:133 conn.rollback()134 print " Catched error (as expected):\n", err135conn.rollback()136curs.execute("DROP TABLE test_copy")137os.unlink('copy_from.txt')...
copy_to.py
Source:copy_to.py
1# copy_to.py -- example about copy_to 2#3# Copyright (C) 2002 Tom Jenkins <tjenkins@devis.com>4# Copyright (C) 2005 Federico Di Gregorio <fog@initd.org>5#6# This program is free software; you can redistribute it and/or modify7# it under the terms of the GNU General Public License as published by the8# Free Software Foundation; either version 2, or (at your option) any later9# version.10#11# This program is distributed in the hope that it will be useful, but12# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTIBILITY13# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License14# for more details.15#16## put in DSN your DSN string17DSN = 'dbname=test'18## don't modify anything below tis line (except for experimenting)19import sys20import os21import StringIO22import psycoev23if len(sys.argv) > 1:24 DSN = sys.argv[1]25print "Opening connection using dns:", DSN26conn = psycoev.connect(DSN)27print "Encoding for this connection is", conn.encoding28curs = conn.cursor()29try:30 curs.execute("CREATE TABLE test_copy (fld1 text, fld2 text, fld3 int4)")31except:32 conn.rollback()33 curs.execute("DROP TABLE test_copy")34 curs.execute("CREATE TABLE test_copy (fld1 text, fld2 text, fld3 int4)")35conn.commit()36# demostrate copy_to functionality37data = [('Tom', 'Jenkins', '37'),38 ('Madonna', None, '45'),39 ('Federico', 'Di Gregorio', None)]40query = "INSERT INTO test_copy VALUES (%s, %s, %s)"41curs.executemany(query, data)42conn.commit()43# copy_to using defaults44io = open('copy_to.txt', 'w')45curs.copy_to(io, 'test_copy')46print "1) Copy %d records into file object using defaults: " % len (data) + \47 "sep = \\t and null = \\N"48io.close()49rows = open('copy_to.txt', 'r').readlines()50print " File has %d rows:" % len(rows)51for r in rows:52 print " ", r,53# copy_to using custom separator54io = open('copy_to.txt', 'w')55curs.copy_to(io, 'test_copy', ':')56print "2) Copy %d records into file object using sep = :" % len(data)57io.close()58rows = open('copy_to.txt', 'r').readlines()59print " File has %d rows:" % len(rows)60for r in rows:61 print " ", r,62# copy_to using custom null identifier63io = open('copy_to.txt', 'w')64curs.copy_to(io, 'test_copy', null='NULL')65print "3) Copy %d records into file object using null = NULL" % len(data)66io.close()67rows = open('copy_to.txt', 'r').readlines()68print " File has %d rows:" % len(rows)69for r in rows:70 print " ", r,71# copy_to using custom separator and null identifier72io = open('copy_to.txt', 'w')73curs.copy_to(io, 'test_copy', ':', 'NULL')74print "4) Copy %d records into file object using sep = : and null ) NULL" % \75 len(data)76io.close()77rows = open('copy_to.txt', 'r').readlines()78print " File has %d rows:" % len(rows)79for r in rows:80 print " ", r,81curs.execute("DROP TABLE test_copy")82os.unlink('copy_to.txt')...
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!!