Code & News

Работа с базой данных PostgreSQL на Python psycopg2

GetCoder.ru
Изображение статьи

В этой статье мы рассмотрим вариант создания класса с методами для подключения и работой с базой данных PostgreSQL. Данный класс поможет нам упростить процесс подключения к базе данных и ускорит процесс написания запросов.

В этой статье мы рассмотрим вариант создания класса с методами для подключения и работой с базой данных PostgreSQL. Данный класс поможет нам упростить процесс подключения к базе данных и ускорит процесс написания запросов.

Описание методов

Наш файл содержит два класса. Давайте кратко рассмотрим содержимое класса dbPostgres, который содержит интересующие нас методы.

  • __init__: Открытие соединения;
  • create_table: Создание таблицы;
  • insert: Внесение данных построчно;
  • select: Вывод данных в виде ассоциативного массива;
  • update: Обновление данных;
  • delete: Удаление данных;
  • truncate: Очистка таблицы;
  • disconnect: Закрытие соединения;

В файле также есть второй класс main, который представлен в виде тестовых запросов для каждого метода.

Установка

Перед тем как начать работу, убедитесь, что у вас установлен модуль psycopg2.

  • copy
pip install psycopg2

Подключение к базе данных

Для подключения к бд требуется установленная база данных и креды доступа.

  • copy

# Параметры подключения
host = "localhost"
port = "5432"
dbname = "test_prod"
user = "postgres"
password = "postgres"
  • host: IP сервера на котором установлена база данных;
  • port: Порт, по которому можно подключиться к базе данных. По умолчанию 5432;
  • dbname: Название базы данных;
  • user: Пользователь в PostgreSQL;
  • password: Пароль от пользователя PostgreSQL;

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

Код

  • copy

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

import psycopg2
from psycopg2.extras import RealDictCursor
import os

class dbPostgres():
	# Подключение
	def __init__(self, host, port, dbname, user, password):
		try:
			self.connection = psycopg2.connect(host = host, port = port, dbname = dbname, user = user, password = password)
			self.cursor = self.connection.cursor(cursor_factory=RealDictCursor)
		except Exception as e:
			exit(str(e))
	
	
	# Создание таблицы
	def create_table(self, nameTable, structure):
		try:
			sql = 'CREATE TABLE IF NOT EXISTS '+nameTable+' ('+structure+')'
			self.cursor.execute(sql)
			self.connection.commit()
			result = True
		except Exception as e:
			result = False
		
		return result
	
	
	# Внесение строки
	def insert(self, variable, nameTable, rows = None, values = None):
		sql = 'INSERT INTO '+nameTable
		sql += ' ('+rows+')' if rows != None else ''
		sql += ' VALUES ('+values+')'
		sql += ' RETURNING id;'
		variable = {} if variable == None else variable
		
		try:
			self.cursor.execute(sql, variable)
			self.connection.commit()
			result = self.cursor.fetchall()[0]
		except Exception as e:
			result = False
		
		return result
	
	
	# Вывод	данных
	def select(self, variable, what, nameTable, where = None, order = None):
		try:
			sql = 'SELECT '+what+' FROM '+nameTable
			sql += ' WHERE '+where if where != None else ''
			sql += ' ORDER BY '+order if order != None else ''
			variable = [] if variable == None else variable
			
			self.cursor.execute(sql, variable)
			result = self.cursor.fetchall()
		except Exception as e:
			result = False
		
		return result
	
	
	# Обновление строки
	def update(self, variable, nameTable, params, where = None):
		try:
			sql = 'UPDATE '+nameTable+' SET '+params
			sql += ' WHERE '+where if where != None else ''
			variable = {} if variable == None else variable
			
			self.cursor.execute(sql, variable)
			self.connection.commit()
			result = True
		except Exception as e:
			result = False
		
		return result
	
	
	# Удаление строки
	def delete(self, variable, nameTable, where):
		try:
			sql = 'DELETE FROM '+nameTable+' WHERE '+where
			variable = {} if variable == None else variable
			self.cursor.execute(sql, variable)
			self.connection.commit()
			result = True
		except Exception as e:
			result = False
		
		return result
	
	
	# Очистка таблицы
	def truncate(self, nameTable):
		try:
			sql = 'TRUNCATE TABLE '+nameTable+' RESTART IDENTITY;'
			self.cursor.execute(sql)
			self.connection.commit()
			result = True
		except Exception as e:
			result = False
		
		return result
	
	
	# Закрыть соединение
	def disconnect(self):
		try:
			if self.cursor:
				self.cursor.close()
			if self.connection:
				self.connection.close()
			result = True
		except Exception as e:
			result = False
		
		return result


# Пример запросов	
class main():
	def __init__(self):
		# Параметры подключения
		host = "localhost"
		port = "5432"
		dbname = "test_prod"
		user = "postgres"
		password = "postgres"
		
		# Подключение
		self.db = dbPostgres(host, port, dbname, user, password)
		
		# CREATE TABLE
		self.db.create_table("table_test", "id SERIAL PRIMARY KEY, name VARCHAR(100), age INT, department VARCHAR(50), salary DECIMAL(10, 2)")
		print("Таблица успешно создана")
		
		# INSERT
		variable = {"name": "Ivan Ivanov", "age": 30, "department": "Engineering", "salary": 55000.00}
		answer = self.db.insert(variable, "table_test", "name, age, department, salary", "%(name)s, %(age)s, %(department)s, %(salary)s")
		print(f"Данные успешно вставлены, id вставленной строки: {answer['id']}")
		
		# SELECT
		variable = {"name": "Ivan Ivanov"}
		answer = self.db.select(variable, '*', "table_test", "name = %(name)s")
		print(f"Запись успешно найдена: {answer[0]['name']}")
		
		# UPDATE
		variable = {"name": "Ivan Ivanov", "age": 40}
		answer = self.db.update(variable, "table_test", "age = %(age)s", "name = %(name)s")
		print("Запись успешно обновлена")
		
		# DELETE
		variable = {"name": "Ivan Ivanov"}
		answer = self.db.delete(variable, "table_test", "name = %(name)s")
		print("Запись успешно удалена")
		
		# TRUNCATE
		answer = self.db.truncate("table_test")
		print("Таблица успешно очищена")
		
		# Закрыть соединение
		self.db.disconnect()


if __name__ == "__main__":
	main()
Исходный код

Заключение

Этот класс демонстрирует, как можно упростить и ускорить время написания запросов в базу данных PostgreSQL на Python используя библиотеку psycopg2.

  • 15.01.2025
  • 42
  • 0

Работа с базой данных PostgreSQL на Python psycopg2

В этой статье мы рассмотрим вариант создания класса с методами для подключения и работой с базой данных PostgreSQL. Данный класс поможет нам упростить процесс подключения к базе данных и ускорит процесс написания запросов.

Описание методов

Наш файл содержит два класса. Давайте кратко рассмотрим содержимое класса dbPostgres, который содержит интересующие нас методы.

  • __init__: Открытие соединения;
  • create_table: Создание таблицы;
  • insert: Внесение данных построчно;
  • select: Вывод данных в виде ассоциативного массива;
  • update: Обновление данных;
  • delete: Удаление данных;
  • truncate: Очистка таблицы;
  • disconnect: Закрытие соединения;

В файле также есть второй класс main, который представлен в виде тестовых запросов для каждого метода.

Установка

Перед тем как начать работу, убедитесь, что у вас установлен модуль psycopg2.

  • copy
pip install psycopg2

Подключение к базе данных

Для подключения к бд требуется установленная база данных и креды доступа.

  • copy

# Параметры подключения
host = "localhost"
port = "5432"
dbname = "test_prod"
user = "postgres"
password = "postgres"
  • host: IP сервера на котором установлена база данных;
  • port: Порт, по которому можно подключиться к базе данных. По умолчанию 5432;
  • dbname: Название базы данных;
  • user: Пользователь в PostgreSQL;
  • password: Пароль от пользователя PostgreSQL;

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

Код

  • copy

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

import psycopg2
from psycopg2.extras import RealDictCursor
import os

class dbPostgres():
	# Подключение
	def __init__(self, host, port, dbname, user, password):
		try:
			self.connection = psycopg2.connect(host = host, port = port, dbname = dbname, user = user, password = password)
			self.cursor = self.connection.cursor(cursor_factory=RealDictCursor)
		except Exception as e:
			exit(str(e))
	
	
	# Создание таблицы
	def create_table(self, nameTable, structure):
		try:
			sql = 'CREATE TABLE IF NOT EXISTS '+nameTable+' ('+structure+')'
			self.cursor.execute(sql)
			self.connection.commit()
			result = True
		except Exception as e:
			result = False
		
		return result
	
	
	# Внесение строки
	def insert(self, variable, nameTable, rows = None, values = None):
		sql = 'INSERT INTO '+nameTable
		sql += ' ('+rows+')' if rows != None else ''
		sql += ' VALUES ('+values+')'
		sql += ' RETURNING id;'
		variable = {} if variable == None else variable
		
		try:
			self.cursor.execute(sql, variable)
			self.connection.commit()
			result = self.cursor.fetchall()[0]
		except Exception as e:
			result = False
		
		return result
	
	
	# Вывод	данных
	def select(self, variable, what, nameTable, where = None, order = None):
		try:
			sql = 'SELECT '+what+' FROM '+nameTable
			sql += ' WHERE '+where if where != None else ''
			sql += ' ORDER BY '+order if order != None else ''
			variable = [] if variable == None else variable
			
			self.cursor.execute(sql, variable)
			result = self.cursor.fetchall()
		except Exception as e:
			result = False
		
		return result
	
	
	# Обновление строки
	def update(self, variable, nameTable, params, where = None):
		try:
			sql = 'UPDATE '+nameTable+' SET '+params
			sql += ' WHERE '+where if where != None else ''
			variable = {} if variable == None else variable
			
			self.cursor.execute(sql, variable)
			self.connection.commit()
			result = True
		except Exception as e:
			result = False
		
		return result
	
	
	# Удаление строки
	def delete(self, variable, nameTable, where):
		try:
			sql = 'DELETE FROM '+nameTable+' WHERE '+where
			variable = {} if variable == None else variable
			self.cursor.execute(sql, variable)
			self.connection.commit()
			result = True
		except Exception as e:
			result = False
		
		return result
	
	
	# Очистка таблицы
	def truncate(self, nameTable):
		try:
			sql = 'TRUNCATE TABLE '+nameTable+' RESTART IDENTITY;'
			self.cursor.execute(sql)
			self.connection.commit()
			result = True
		except Exception as e:
			result = False
		
		return result
	
	
	# Закрыть соединение
	def disconnect(self):
		try:
			if self.cursor:
				self.cursor.close()
			if self.connection:
				self.connection.close()
			result = True
		except Exception as e:
			result = False
		
		return result


# Пример запросов	
class main():
	def __init__(self):
		# Параметры подключения
		host = "localhost"
		port = "5432"
		dbname = "test_prod"
		user = "postgres"
		password = "postgres"
		
		# Подключение
		self.db = dbPostgres(host, port, dbname, user, password)
		
		# CREATE TABLE
		self.db.create_table("table_test", "id SERIAL PRIMARY KEY, name VARCHAR(100), age INT, department VARCHAR(50), salary DECIMAL(10, 2)")
		print("Таблица успешно создана")
		
		# INSERT
		variable = {"name": "Ivan Ivanov", "age": 30, "department": "Engineering", "salary": 55000.00}
		answer = self.db.insert(variable, "table_test", "name, age, department, salary", "%(name)s, %(age)s, %(department)s, %(salary)s")
		print(f"Данные успешно вставлены, id вставленной строки: {answer['id']}")
		
		# SELECT
		variable = {"name": "Ivan Ivanov"}
		answer = self.db.select(variable, '*', "table_test", "name = %(name)s")
		print(f"Запись успешно найдена: {answer[0]['name']}")
		
		# UPDATE
		variable = {"name": "Ivan Ivanov", "age": 40}
		answer = self.db.update(variable, "table_test", "age = %(age)s", "name = %(name)s")
		print("Запись успешно обновлена")
		
		# DELETE
		variable = {"name": "Ivan Ivanov"}
		answer = self.db.delete(variable, "table_test", "name = %(name)s")
		print("Запись успешно удалена")
		
		# TRUNCATE
		answer = self.db.truncate("table_test")
		print("Таблица успешно очищена")
		
		# Закрыть соединение
		self.db.disconnect()


if __name__ == "__main__":
	main()
Исходный код

Заключение

Этот класс демонстрирует, как можно упростить и ускорить время написания запросов в базу данных PostgreSQL на Python используя библиотеку psycopg2.