07.03.2023 в 15:55

Как подключить Python и PostgreSQL

В этой статье мы рассмотрим краткий пример того, как мы можем взаимодействовать с нашей базой данных PostgreSQL через Python через psycopg2 API.

Прежде чем мы начнем:

Убедитесь, что у вас установлен PostgreSQL
Убедитесь, что вы установили psycopg2: pip install psycopg2-binary

Сначала давайте сохраним учетные данные нашей базы данных и проанализируем их в Python, чтобы мы могли подключиться:

Наш файл конфигурации:

# database.ini (Make sure to git ignore this!!!)
[postgresql]
host=localhost
database=your_database_name
user=postgres
password=postgres

Наш код на данный момент:

import psycopg2
from configparser import ConfigParser

class PSQL:
    connection=None
    cursor=None

    # Makes inital connection to our database or throws an exception
    def __init__(self):
        # Parsing database.ini credentials for connection
        parser=ConfigParser()
        parser.read('database.ini')

        db={}
        params=parser.items('postgresql')
        for param in params:
            db[param[0]] = param[1]

В этом фрагменте мы используем ConfigParser() для чтения нашего файла database.ini, а затем прокручиваем и сохраняем его в dict.

Теперь давайте установим соединение с нашей базой данных:

import psycopg2
from configparser import ConfigParser

class PSQL:
    connection=None
    cursor=None

    # Makes inital connection to our database or throws an exception
    def __init__(self):
        # Parsing database.ini credentials for connection
        parser=ConfigParser()
        parser.read('database.ini')

        db={}
        params=parser.items('postgresql')
        for param in params:
            db[param[0]] = param[1]

        # Connect to database, feel free to make error handling here
        conn = psycopg2.connect(**db)
        PSQL.connection = conn
        PSQL.cursor = conn.cursor()

Это в некоторой степени просто, и на этом этапе вы можете добавить любой вид обработки ошибок, который пожелаете. Я также делаю это объектно-ориентированным способом, однако вы могли бы сделать это функционально, если бы захотели!

Наконец, давайте напишем метод для взаимодействия с нашей базой данных!

import psycopg2
from configparser import ConfigParser

class PSQL:
    connection=None
    cursor=None

    # Makes inital connection to our database or throws an exception
    def __init__(self):
        # Parsing database.ini credentials for connection
        parser=ConfigParser()
        parser.read('database.ini')

        db={}
        params=parser.items('postgresql')
        for param in params:
            db[param[0]] = param[1]

        # Connect to database, feel free to make error handling here
        conn = psycopg2.connect(**db)
        PSQL.connection = conn
        PSQL.cursor = conn.cursor()

    def query(self, string, params={}):
        result=None

        # Perform the query
        if params:
            PSQL.cursor.execute(string, params)
        else:
            PSQL.cursor.execute(string)

        # Retrieve the result (only if the query returned one)
        if PSQL.cursor.description is not None:
            result = PSQL.cursor.fetchone()

        # Commit changes
        PSQL.connection.commit()
        return result

Это наш основной метод запроса! Добавим несколько бонусных внизу, и вы должны быть готовы перейти к этому пункту. Примечание: при извлечении вы должны убедиться, что запрос действительно что-то вернул, в противном случае будет выдано исключение.

Некоторые примеры:

from my_file_name import PSQL
db = PSQL()
# Example of some queries that wouldnt return anything => None
db.query("""CREATE TABLE x (id INT PRIMARY KEY);""")
db.query("""INSERT INTO x (id) VALUES (%s);""", (1,))
# Example of a query that would return something
result = db.query("""SELECT * FROM x WHERE id = %s""", (1,))

Метод бонусной транзакции

Метод запуска транзакции вместо фиксации после каждой

    def transaction(self, queries): # queries => [[query_string, params], ...]
        try:
            query_results = []
            for query in queries:
                statement = query[0]
                params = query[1] ? query[1] : {}
                PSQL.cursor.execute(statement, params)
                if PSQL.cursor.description is not None:
                    query_results.append(PSQL.cursor.fetchone())
                else:
                    query_results.append(None)
            PSQL.connection.commit()
            return query_results
        except psycopg2.DatabaseError as error:
            print(error)

#Flask #PostgreSQL #Python

542

Войти на сайт

Как подключить Python и PostgreSQL