-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathPyFuncs4PostgreSQL.py
140 lines (125 loc) · 4.06 KB
/
PyFuncs4PostgreSQL.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
import psycopg2
import configparser
from osgeo import ogr
import numpy as np
import sys
import os
# os.chdir(os.path.dirname(sys.argv[0]))
#create a dictionary of parameter using configparser
def config (configFile = r'db.txt', section = 'postgresql'):
parserA = configparser.ConfigParser()
parserA.read(configFile)
return dict(parserA.items(section))
def insertSQL (db,table,fieldsValues):
conn = psycopg2.connect(**db)
cur = conn.cursor()
fieldState =str()
valueState =str()
for field, value in fieldsValues.items():
fieldState = fieldState + "%s, "%(field)
valueState = valueState + "'%s', "%(value)
state = "INSERT INTO " + table + ' (' + fieldState[0:-2] + ') VALUES (' + valueState[0:-2] + ')'
print (state)
cur.execute(state)
conn.commit()
conn.close()
def selectAll (db,table):
conn = psycopg2.connect(**db)
cur = conn.cursor()
cur.execute("SELECT * FROM %s"%(table))
records = cur.fetchall()
conn.close()
return records
def selectCol_where (db,table, *fields, where):
conn = psycopg2.connect(**db)
cur = conn.cursor()
fieldState =str()
for field in fields:
fieldState = fieldState + "%s, "%(field)
state = 'SELECT ' + fieldState[0:-2] + ' FROM ' + table + ' WHERE ' + where
print (state)
cur.execute(state)
records = cur.fetchall()
conn.close()
return records
def inner_join(db,tab1, tab2, connector,*fields,where):
conn = psycopg2.connect(**db)
cur = conn.cursor()
fieldState =str()
for field in fields:
fieldState = fieldState + "%s, "%(field)
state = 'SELECT %s FROM %s INNER JOIN %s ON %s.%s=%s.%s WHERE %s'%(fieldState[0:-2],tab1, tab2, tab1, connector, tab2 , connector, where)
print (state)
cur.execute(state)
records = cur.fetchall()
conn.close()
return records
def selectCol (db,table, *fields):
conn = psycopg2.connect(**db)
cur = conn.cursor()
fieldState =str()
for field in fields:
fieldState = fieldState + "%s, "%(field)
state = 'SELECT ' + fieldState[0:-2] + ' FROM ' + table
cur.execute(state)
records = cur.fetchall()
conn.close()
return records
#usage: update ('hehe',*{a = 1, b = 2}, where = "he = 'him'")
#or: update ('hehe',a = 1, b = 2, where = "he = 'him'")
def update (db,table,where,**fieldsValues):
conn = psycopg2.connect(**db)
cur = conn.cursor()
fieldValueState = str()
for field, value in fieldsValues.items():
fieldValueState = fieldValueState + field + ' = ' + "'%s', " %(value)
state = 'UPDATE ' + table + ' SET ' + fieldValueState[0:-2] +' WHERE ' + where
cur.execute(state)
conn.commit()
conn.close()
def getGeom(db,Xmin,Ymin,Xmax,Ymax,Projection):
conn = psycopg2.connect(**db)
cur = conn.cursor()
state = 'SELECT ST_MakeEnvelope('+ str(Xmin) +',' + str(Ymin) +','+ str(Xmax) +','+ str(Ymax) +','+ str(Projection) +')'
cur.execute(state)
records = cur.fetchall()
conn.close()
return records[0][0]
def getCoord(db,geom):
conn = psycopg2.connect(**db)
cur = conn.cursor()
state = 'SELECT ST_AsText('+ "'%s'"%(geom) +')'
cur.execute(state)
records = cur.fetchall()
conn.close()
return records[0][0]
table = 'cctl_giatri_doman'
fields = ['maso_tramdo','giatri']
where = "thoigian = '%s'"%(date)
station_value = selectCol (table, *fields, where = where)
print (station_value)
values, stations, xcoords, ycoords = list(), list(), list(), list()
def select_row_json (db,tab1, tab2, connector,*fields,where):
conn = psycopg2.connect(**db)
cur = conn.cursor()
fieldState =str()
for field in fields:
fieldState = fieldState + "%s, "%(field)
state = 'SELECT row_to_json(%s) FROM %s INNER JOIN %s ON %s.%s=%s.%s WHERE %s'%(tab1,tab1, tab2, tab1, connector, tab2, connector, where)
print (state)
cur.execute(state)
records = cur.fetchall()
conn.close()
return records
def select_ST_AsGeoJSON (db,tab1, tab2, connector,*fields,where):
conn = psycopg2.connect(**db)
cur = conn.cursor()
fieldState =str()
for field in fields:
fieldState = fieldState + "%s, "%(field)
state = 'SELECT ST_AsGeoJSON(%s) FROM %s INNER JOIN %s ON %s.%s=%s.%s WHERE %s'%(*fields,tab1, tab2, tab1, connector, tab2, connector, where)
print (state)
cur.execute(state)
records = cur.fetchall()
conn.close()
return records