Python操作数据库(mysql、pgsql)

记录一些python直接操作数据库得代码示例

mysql

1. python3 利用pymysql封装sql操作(非ORM)
2. 数据返回json格式

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# __author__ = 'Ysc'
import pymysql
from base_config import DB_HOST, DB_USER, DB_DATABASE, DB_PASSWORD, DB_PORT


class MysqlHelp():
    """构造
    def __init__(self, host=DB_HOST, user=DB_USER, port=DB_PORT, password=DB_PASSWORD, database=DB_DATABASE):
        self.host = host
        self.user = user
        self.port = port
        self.password = password
        self.db = database
    """
    # 创建连接
    def open_coon(self,host=DB_HOST, user=DB_USER, port=DB_PORT, password=DB_PASSWORD, database=DB_DATABASE):
        self.coon = pymysql.connect(host=host, port=port, user=user, passwd=password, db=database)
        self.cursor = self.coon.cursor(cursor=pymysql.cursors.DictCursor)

    # 关闭连接
    def close(self):
        self.cursor.close()
        self.coon.cursor()

    # 调用语句
    def insert_delete_update(self, sql, params=[]):
        try:
            # self.open_coon()

            self.cursor.execute(sql, params)
            self.coon.commit()

            # self.close()

            return "OK"
        except Exception as erorr:
            return erorr

    # 查询 接收全部的返回结果行

    def select_fetchall(self, sql, params=[]):
        try:
            # self.open_coon()

            self.cursor.execute(sql, params)

            results = self.cursor.fetchall()

            self.coon.commit()

            # self.close()
            return results

        except Exception as erorr:
            return erorr

使用ssh操作pgsql

以下操作也可以用来 链接 Mysql

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# __author__ = 'Ysc'

import psycopg2
import psycopg2.extras
from sshtunnel import SSHTunnelForwarder
import pandas as pd

server = SSHTunnelForwarder(
    # 指定ssh登录的跳转机的address
    ssh_address_or_host=('ip', 22),
    ssh_username='root',
    # 设置密钥
    ssh_pkey='',
    ssh_private_key_password='',

    # 设置数据库服务地址及端口
    remote_bind_address=('***', 5432))
server.start()
conn = psycopg2.connect(database='kong',
                        user='admin',
                        password='****',
                        host='127.0.0.1',
                        port=server.local_bind_port)
# cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
sql = 'SELECT * FROM "public"."oauth2_tokens" LIMIT 1000 OFFSET 0'
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cur.execute(sql)
rows = cur.fetchall()


# df = pd.read_sql(sql, conn)
conn.close()
print(rows,type(rows))
server.close()


#
# conn = psycopg2.connect(database="testdb", user="postgres", password="pass123", host="127.0.0.1", port="5432")