pg_db_func.py
3.5 KB
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
#! /usr/bin/env python
# -*- coding: utf-8 -*-
# Date: 2020/7/12 16:27
# @Author:bxh
# @file: pg_db_func.py
import psycopg2
# 获得连接
def GetConnect():
conn = False
try:
conn = psycopg2.connect(database="partner_test",
user="bianxinhua_user",
password="Bianxinhua_user#123",
host="101.37.68.23",
port="15432")
except Exception as err:
print("连接数据库失败，%s" % err)
return conn
def get_id_from_database(tbl):
db = GetConnect()
cursor = db.cursor()
sql= "select id from " + tbl + ' ' + 'order by id desc '
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
def get_account_from_database(tbl):
db = GetConnect()
cursor = db.cursor()
sql= "select account from " + tbl + ' ' + 'order by id desc '
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
def get_partnername_from_database(tbl):
db = GetConnect()
cursor = db.cursor()
sql= "select partner_name from " + tbl + ' ' + 'order by id desc '
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
def get_area_from_database(tbl):
db = GetConnect()
cursor = db.cursor()
sql= "select region_info from " + tbl + ' ' + 'order by id desc '
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]['regionName']
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
def get_salesman_from_database(tbl):
db = GetConnect()
cursor = db.cursor()
sql= "select salesman from " + tbl + ' ' + 'order by id desc '
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0][0]['name']
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
def get_phone_from_database(tbl):
db = GetConnect()
cursor = db.cursor()
sql= "select salesman from " + tbl + ' ' + 'order by id desc '
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0][0]['telephone']
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
def get_status_from_database(tbl):
db = GetConnect()
cursor = db.cursor()
sql= "select status from " + tbl + ' ' + 'order by id desc '
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
def get_partnername_from_database(tbl):
db = GetConnect()
cursor = db.cursor()
sql= "select partner_name from " + tbl + ' ' + "where partner_category = 1" + ' ' + 'order by id desc '
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
print(get_id_from_database('partner_info'))