定时插入数据测试脚本

栏目: 数据库 · 发布时间: 5年前

内容简介:import cx_Oracleclass Oracle(object):

insert.py

import cx_Oracle

class Oracle(object):

“”" oracle db operator “”"

def init (self,userName,password,host,instance):

self._conn = cx_Oracle.connect("%s/%s@%s/%s" % (userName,password,host,instance))

self.cursor = self._conn.cursor()

def queryTitle(self,sql,nameParams={}):

if len(nameParams) > 0 :

self.cursor.execute(sql,nameParams)

else:

self.cursor.execute(sql)

colNames = []

for i in range(0,len(self.cursor.description)):

colNames.append(self.cursor.description[i][0])

return colNames

# query methods

def queryAll(self,sql):

self.cursor.execute(sql)

return self.cursor.fetchall()

def queryOne(self,sql):

self.cursor.execute(sql)

return self.cursor.fetchone()

def queryBy(self,sql,nameParams={}):

if len(nameParams) > 0 :

self.cursor.execute(sql,nameParams)

else:

self.cursor.execute(sql)

return self.cursor.fetchall()

def insertBatch(self,sql,nameParams=[]):

“”“batch insert much rows one time,use location parameter”""

self.cursor.prepare(sql)

self.cursor.executemany(None, nameParams)

self.commit()

def commit(self):

self._conn.commit()

def del (self):

if hasattr(self,‘cursor’):

self.cursor.close()

if hasattr(self,’_conn’):

self.

conn.close()

def test1():

# sql = “”“select user_name,user_real_name,to_char(create_date,‘yyyy-mm-dd’) create_date from sys_user where id = ‘10000’ “””

sql = “”“select user_name,user_real_name,to_char(create_date,‘yyyy-mm-dd’) create_date from sys_user where id =: id “””

oraDb = Oracle(‘test’,‘java’,‘192.168.0.192’,‘orcl’)

fields = oraDb.queryTitle(sql, {‘id’:‘10000’})

print(fields)

print(oraDb.queryBy(sql, {‘id’:‘10000’}))

def test2():

oraDb = Oracle(‘test’,‘test’,‘129.154.102.100’,‘PDB1.612739963.oraclecloud.internal’)

cursor = oraDb.cursor

#drop_table=""“drop table python_modules”""

#cursor.execute(drop_table)

import random

"""+str(random.randint(1,4000))

create_table = “”"

CREATE TABLE “”"+str(lat)+""" (

module_name VARCHAR2(50) NOT NULL,

file_path VARCHAR2(300) NOT NULL

)

“”"

from sys import modules

cursor.execute(create_table)

M = []

for m_name, m_info in modules.items():

try:

M.append((m_name, m_info. file ))

except AttributeError:

pass

sql = “INSERT INTO “+lat+”(module_name, file_path) VALUES (:1, :2)”

oraDb.insertBatch(sql,M)

cursor.execute("SELECT COUNT(*) FROM "+lat)

print(cursor.fetchone())

print(‘insert batch ok.’)

#cursor.execute(“DROP TABLE python_modules PURGE”)

insert.sh

python insert.py

run.sh

#!/bin/bash

#需要定时执行的程序

program=./insert.sh

#获取当前时间,例如20171129 <—> perDate= KaTeX parse error: Expected 'EOF', got '#' at position 18: …ate "+%Y%m%d") #̲获取一天后的时间 afterd… (date -d +1day “+%Y%m%d”)

#afterdata=20170323

#每五分钟仅仅执行一次,设置标志位

onceflag=0

echo ‘Task schedule Time: (’ KaTeX parse error: Expected 'EOF', got '#' at position 53: … per 5min)...' #̲死循环,当检测到当前时间段分钟… (date “+%M”);

if [ expr $presentminutes % 120 -eq 0 ]

then

if [ o n c e f l a g e q 0 ] t h e n e c h o N o w m i n u t e i s onceflag -eq 0 ] then echo &#x27;Now minute is &#x27; presentminutes’’

echo ‘The program (’$program’) running…’

echo $program $afterdata #打印测试可执行程序格式是否正确

$program a f t e r d a t a e c h o T h e p r o g r a m ( afterdata echo &#x27;The program (&#x27; program’) stopped…’

onceflag=1

else

continue

fi

else

onceflag=0

fi

sleep 1 #一秒查询一次当前时间点

done


以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,也希望大家多多支持 码农网

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

The Shallows

The Shallows

Nicholas Carr / W. W. Norton & Company / 2010-6-15 / USD 26.95

"Is Google making us stupid?" When Nicholas Carr posed that question, in a celebrated Atlantic Monthly cover story, he tapped into a well of anxiety about how the Internet is changing us. He also crys......一起来看看 《The Shallows》 这本书的介绍吧!

Base64 编码/解码
Base64 编码/解码

Base64 编码/解码

正则表达式在线测试
正则表达式在线测试

正则表达式在线测试

RGB HSV 转换
RGB HSV 转换

RGB HSV 互转工具