Localization-onemap-implementationSupervision-system-python/db/postgresql.py

# -*- coding: utf-8 -*-
import decimal
# 设置语言环境
import os
import sys
import json
from datetime import datetime

import log
import importlib, sys

importlib.reload(sys)

os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.UTF8'

import psycopg2


class Postgresql:
    def __init__(self, conn):
        """Postgresql
        C:\Program Files\GeoScene\Pro\bin\Python\envs\arcgispro-py3\Scripts
        pip install psycopg2
        :param conn：数据库连接字符串（'user/password@127.0.0.1/orcl'）
        """
        # self._conn = psycopg2.connect(**conn, encoding="UTF-8", nencoding="UTF-8")
        self._conn = psycopg2.connect(**conn)
        self.cursor = self._conn.cursor()  # 游标

    def commit(self):
        self._conn.commit()

    def close(self):
        try:
            if hasattr(self, 'cursor'):
                self.cursor.close()
            if hasattr(self, '_conn'):
                self._conn.close()
        except:
            msg = str(sys.exc_info()).decode('string-escape')
            log.error(msg)

    def __del__(self):
        self.close()

    def query(self, sql):
        """查询所有数据
        :param sql：SQL语句
        """
        self.cursor.execute(sql)
        ds = self.cursor.fetchall()
        cols = [d[0] for d in self.cursor.description]
        list = []
        for row in ds:
            b = dict(zip(cols, row))
            list.append(b)
        jobj = json.loads(json.dumps(list, cls=self.CustomJsonEncoder, ensure_ascii=False))
        return jobj

    class CustomJsonEncoder(json.JSONEncoder):
        """
        Json解析器，解决识别Decimal出错的问题
        """

        def default(self, obj):
            if isinstance(obj, datetime):
                return obj.strftime("%Y-%m-%d %H:%M:%S")
            if isinstance(obj, bytes):
                return str(obj, encoding='utf-8')
            if isinstance(obj, int):
                return int(obj)
            elif isinstance(obj, float):
                return float(obj)
            elif isinstance(obj, decimal.Decimal):
                return float(obj)
            # elif isinstance(obj, array):
            #    return obj.tolist()
            else:
                return super(self.CustomJsonEncoder, self).default(obj)

    def insert(self, sql, dict=None):
        """查询数据
        :param sql：SQL语句 ==> insert into 表(字段1,字段2) values(:字段1,:字段2)
        :param dict：值字典 ==> {"字段1":"abc", "字段2":"abc"}
        """
        if dict == None:
            self.cursor.execute(sql)
        else:
            self.cursor.execute(sql, dict)
        self.commit()