Paradox DB helper class in Python using pypyodbc

README

Tested on Windows 10 Pro (64bit)
Python version 3.7.4 (32bit)

Note for your mental sanity: the "Microsoft Paradox Driver" was at 32bit only and work ONLY with Python 32bit version!

paradox_helper.py

import os
import re
import shutil
import pypyodbc


class ParadoxHelper:
	"""
	author: Antonio Porcelli
	username: Progressify
	github: https://github.com/progressify
	ig: https://www.instagram.com/progressify/
	site: https://progressify.dev
	special_thanks: https://mediarealm.com.au/articles/accessing-paradox-databases-python/
	"""

	db_connection = None
	db_folder = None
	sql_connection_string = r"Driver={{Microsoft Paradox Driver (*.db )}};DriverID=538;Fil=Paradox 7.X;DefaultDir={0};Dbq={0};CollatingSequence=ASCII;"
	temp_folder = ""
	table_name = ""

	def __init__(self, a_table_name, a_db_folder):
		self.db_folder = a_db_folder
		self.table_name = a_table_name
		self.setup_temp_copy()
		self.connect()

	def setup_temp_copy(self, a_table_name=None, a_db_folder=None):
		if a_db_folder is None: a_db_folder = self.db_folder
		if a_table_name is None: a_table_name = self.table_name

		# Work out the path to our temporary folder
		self.temp_folder = os.path.join(os.path.dirname(os.path.abspath(__file__)), "DBTEMP")
		# Delete the temporary directory (if exists)
		shutil.rmtree(self.temp_folder, True)
		os.makedirs(self.temp_folder, exist_ok=True)
		src_files = os.listdir(self.db_folder)
		for file_name in src_files:
			if re.search(r'^{}+\.([a-zA-Z0-9]*)'.format(a_table_name), file_name):
				shutil.copyfile('{}{}'.format(a_db_folder, file_name), '{}\\{}'.format(self.temp_folder, file_name))

		# Remove existing lock files from the database
		try:
			shutil.move(os.path.join(self.temp_folder, "net", "PDOXUSRS.NET"),
						os.path.join(self.temp_folder, "net", "REMOVED-PDOXUSRS.NET"))
		except:
			pass

		try:
			shutil.move(os.path.join(self.temp_folder, "PDOXUSRS.LCK"), os.path.join(self.temp_folder, "REMOVED-PDOXUSRS.LCK"))
		except:
			pass

		try:
			shutil.move(os.path.join(self.temp_folder, "PARADOX.LCK"), os.path.join(self.temp_folder, "REMOVED-PARADOX.LCK"))
		except:
			pass

		return self.temp_folder

	def connect(self, tmp_folder=None):
		# Setup Paradox SQL DB Connection and return a cursor
		if tmp_folder is None: tmp_folder = self.temp_folder

		self.db_connection = pypyodbc.connect(self.sql_connection_string.format(tmp_folder), autocommit=True)
		return self.db_connection.cursor()

	def query(self, query, cursor=None, params=None):
		# Run a DB Query and return the results as a list of dicts
		if cursor is None: cursor = self.db_connection.cursor()
		if params is None: params = []

		cursor.execute(query, params)
		headers = [item[0] for item in cursor.description]
		returndata = []

		for x in cursor:
			thisrow = {}
			for i, y in enumerate(x):
				thisrow[headers[i]] = y
			returndata.append(thisrow)

		return returndata

	def close(self, cursor=None):
		if cursor is None: self.db_connection.cursor().close()
		else: cursor.cursor()


# usage
if __name__ == "__main__":
	pdox_h = ParadoxHelper("Clienti", "C:\\2019\\")
	print(pdox_h.query('SELECT * FROM Clienti WHERE codice = ?', params=[60072]))
	pdox_h.close()

requirements.txt

pypyodbc==1.3.4