File size: 1,972 Bytes
25d2b3e ae93751 25d2b3e ae93751 25d2b3e ae93751 25d2b3e ae93751 25d2b3e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 |
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
وحدة الاتصال بقاعدة البيانات
"""
import os
import sys
import psycopg2
from dotenv import load_dotenv
# إضافة مسار النظام للوصول للملفات المشتركة
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
# تحميل متغيرات البيئة
load_dotenv()
def get_connection():
"""
إنشاء اتصال بقاعدة البيانات
الإرجاع:
اتصال بقاعدة البيانات
"""
try:
# محاولة الاتصال بقاعدة البيانات
conn = psycopg2.connect(
dbname=os.getenv("PGDATABASE"),
user=os.getenv("PGUSER"),
password=os.getenv("PGPASSWORD"),
host=os.getenv("PGHOST"),
port=os.getenv("PGPORT")
)
return conn
except Exception as e:
print(f"خطأ في الاتصال بقاعدة البيانات: {e}")
# إذا فشل الاتصال، استخدم اتصال قاعدة بيانات SQLite محلية
import os
import sqlite3
# إنشاء مجلد البيانات إذا لم يكن موجوداً
data_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'data')
os.makedirs(data_dir, exist_ok=True)
# إنشاء اتصال قاعدة بيانات SQLite محلية
db_path = os.path.join(data_dir, 'local_db.sqlite')
conn = sqlite3.connect(db_path)
# إعادة محاكاة سلوك اتصال PostgreSQL
conn.execute = conn.cursor().execute
# إضافة وظيفة وهمية للاقتطاع (commit) والإغلاق
original_close = conn.close
def enhanced_close():
conn.commit()
original_close()
conn.close = enhanced_close
return conn |