pg_db_func.py
3.9 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
133
134
135
136
137
138
139
140
141
#! /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_admin_account_from_database(tbl):
db = GetConnect()
cursor = db.cursor()
# sql= "select id from " + '''"''' + tbl + '''"'''+ ' ' + 'order by id desc '
sql = '''select account from \"{}\" where is_default ='t' '''.format(tbl)
# print(sql)
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#实际订单的发货单号
def get_actual_delivery_code_from_database():
db = GetConnect()
cursor = db.cursor()
sql = '''select delivery_code from order_base where order_type = 1 order by id DESC'''
#print(sql)
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#实际订单的id
def get_actual_id_from_database():
db = GetConnect()
cursor = db.cursor()
sql = '''select id from order_base where order_type = 1 order by id DESC'''
#print(sql)
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#意向订单id
def get_purpose_id_from_database():
db = GetConnect()
cursor = db.cursor()
# sql= "select id from " + '''"''' + tbl + '''"'''+ ' ' + 'order by id desc '
sql = '''select id from order_base where order_type = 2 order by id DESC'''
try:
cursor.execute(sql)
result = cursor.fetchone()
return result[0]
except Exception as err:
# 发生错误时回滚
print(err)
db.rollback()
db.close()
#意向订单的产品ID
def get_ID_from_database(id):
db = GetConnect()
cursor = db.cursor()
sql= '''select id from order_good where order_id = {} order by id ASC '''.format(get_purpose_id_from_database())
# print(sql)
try:
cursor.execute(sql)
result = cursor.fetchall()
return result[id][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_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()