sanya-dbms/processing/tools/Login/PostgreSQL.py

# -*- coding: utf-8 -*-

__author__ = 'wanger'
__date__ = '2024-08-20'
__copyright__ = '(C) 2024 by siwei'
__revision__ = '1.0'

import time
from typing import Optional
import os
import psycopg2
import uuid
import siwei_config


class PostgreSQL:
    # 用户表
    Sys_user = "sys_user"
    # 全局参数修改
    db_config = siwei_config.CONFIG['db']

    def __init__(
            self,
            host: Optional[str] = db_config['host'],  # default host during installation
            port: Optional[str] = db_config['port'],  # default port during pg installation
            user: Optional[str] = db_config['user'],  # default user during pg installation
            password: Optional[str] = db_config['password'],  # default password during pg installation
            dbname: Optional[str] = db_config['name'],  # default dbname during pg installation
            schema: Optional[str] = db_config['schema']
    ):
        # 配置数据库连接参数并指定schema
        self.connparams = {
            "dbname": dbname,
            "user": user,
            "password": password,
            "host": host,
            "port": port,
            "options": "-c search_path=otherSchema," + schema if schema is not None else None
        }
        self.conn = psycopg2.connect(**self.connparams)
        # 创建一个游标对象
        self.cur = self.conn.cursor()

    def execute(self, sql):
        # 执行一个查询
        self.cur.execute(sql)
        # 获取查询结果
        return self.cur.fetchall()

    def close(self):
        # 关闭游标和连接
        self.cur.close()
        self.conn.close()

    # 根据用户名查询密码
    def getPasswordByUsername(self, username):
        sql = "SELECT password FROM {}  t where t.user_name = '{}'".format(self.Sys_user, username)
        self.cur.execute(sql)
        rows = self.cur.fetchall()
        return rows

    # 判断数据是否为字符串
    def is_string(self, var):
        return isinstance(var, str)