python-module-pgsql/manage_psql.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Module psql connection."""

# /*
# * ----------------------------------------------------------------------------
# * "THE BEER-WARE LICENSE" (Revision 42):
# * <boris.tassou@securmail.fr> wrote this file. As long as you retain this notice you
# * can do whatever you want with this stuff. If we meet some day, and you think
# * this stuff is worth it, you can buy me a beer in return Boris Tassou
# * ----------------------------------------------------------------------------
# */

import socket
from configparser import ConfigParser
import psycopg2

# Read conf.ini
INI_CONF = "./config/psql_conf.ini"
CONFIG = ConfigParser()
CONFIG.read(INI_CONF)

# DB parameters
PSQL_HOST = CONFIG.get('psql', 'host')
PSQL_DB = CONFIG.get('psql', 'database')
PSQL_USER = CONFIG.get('psql', 'user')
PSQL_PWD = CONFIG.get('psql', 'passwd')


class HandlePsql:
    """Class to handle psql connection."""

    conn = None
    cur = None

    @staticmethod
    def psql_checkport():
        """Check pgsql port."""
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        check_port = sock.connect_ex((PSQL_HOST, 5432))
        if check_port:
            return True
        return False

    @staticmethod
    def psql_checkcon():
        """Check pgsql connexion to database."""
        try:
            conn = psycopg2.connect(host=PSQL_HOST, database=PSQL_DB,
                                    user=PSQL_USER, password=PSQL_PWD,
                                    connect_timeout=5)
            conn.close()
            return True
        except conn.error:
            return False

    @staticmethod
    def psql_conn():
        """Init psql connection."""
        # connexion DB
        # http://initd.org/psycopg/docs/usage.html
        # http://www.postgresqltutorial.com/postgresql-python/
        try:
            global conn
            conn = psycopg2.connect(host=PSQL_HOST, database=PSQL_DB,
                                    user=PSQL_USER, password=PSQL_PWD)
            conn.cursor()
        except psycopg2.DatabaseError as error:
            print(error)

    @staticmethod
    def psql_select(sql_select):
        """Psql select request."""
        cur = conn.cursor()
        cur.execute(sql_select)
        sql_return = cur.fetchall()
        return sql_return

    @staticmethod
    def psql_delete(sql_delete):
        """Psql delte request."""
        cur = conn.cursor()
        cur.execute(sql_delete)
        conn.commit()

    @staticmethod
    def psql_insert(sql_insert):
        """Psql insert request."""
        cur = conn.cursor()
        cur.execute(sql_insert)
        conn.commit()

    @staticmethod
    def psql_update(sql_update):
        """Psql insert request."""
        cur = conn.cursor()
        cur.execute(sql_update)
        conn.commit()

    @staticmethod
    def psql_unconn():
        """End psql connection."""
        global conn
        cur = conn.cursor()
        cur.close()
        conn.close()


if __name__ == "__main__":
    print("Module psql to handle the connection")