-
Notifications
You must be signed in to change notification settings - Fork 0
/
database.py
111 lines (94 loc) · 4.27 KB
/
database.py
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
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from contextlib import contextmanager
from sqlalchemy.exc import SQLAlchemyError
import logging
from config import SQLALCHEMY_DATABASE_URI
from models import DeviceDetail
from base import Base
engine = create_engine(SQLALCHEMY_DATABASE_URI)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Session = scoped_session(SessionLocal)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class DatabaseError(Exception):
"""Custom exception for database-related errors."""
pass
def init_db():
Base.metadata.create_all(bind=engine)
def get_all_device_details():
with db_session_scope() as session:
return session.query(DeviceDetail).all()
def validate_device_detail(device_name, status, last_activity):
# This is a generic validation. Adjust as per your requirements.
if not device_name or not status or not last_activity:
return False
return True
@contextmanager
def db_session_scope():
session = Session()
try:
yield session
session.commit()
except SQLAlchemyError as e:
logger.error(f"Database error: {e}")
session.rollback()
raise
finally:
session.close()
def add_device_detail(device_name, status, last_activity):
if not validate_device_detail(device_name, status, last_activity):
return {"status": "error", "message": "Invalid device details"}
try:
with db_session_scope() as session:
device = session.query(DeviceDetail).filter_by(device_name=device_name).first()
if device:
device.status = status
device.last_activity = last_activity
logger.info(f"Updated device detail for {device_name}")
return {"status": "success", "message": "Device detail updated successfully"}
else:
device_detail = DeviceDetail(device_name=device_name, status=status, last_activity=last_activity)
session.add(device_detail)
logger.info(f"Added new device detail for {device_name}")
return {"status": "success", "message": "Device detail added successfully"}
except SQLAlchemyError as e:
logger.error(f"Error in add_device_detail: {e}")
return {"status": "error", "message": f"Database error: {e}"}
def delete_device_detail(device_name):
try:
with db_session_scope() as session:
device = session.query(DeviceDetail).filter_by(device_name=device_name).first()
if device:
session.delete(device)
return {"status": "success", "message": "Device detail deleted successfully"}
else:
return {"status": "error", "message": "Device detail not found"}
except SQLAlchemyError as e:
logger.error(f"Error in delete_device_detail: {e}")
return {"status": "error", "message": f"Database error: {e}"}
def modify_device_detail(device_name, status=None, last_activity=None):
try:
with db_session_scope() as session:
device = session.query(DeviceDetail).filter_by(device_name=device_name).first()
if not device:
device = DeviceDetail(device_name=device_name)
session.add(device)
if status is not None:
device.status = status
if last_activity is not None:
device.last_activity = last_activity
return {"status": "success", "message": "Device detail modified successfully"}
except Exception as e:
logger.error(f"Error in modify_device_detail: {e}")
raise DatabaseError(f"Error while modifying device detail: {e}")
def get_device_detail_by_name(device_name):
try:
with db_session_scope() as session:
device = session.query(DeviceDetail).filter_by(device_name=device_name).first()
return device
except Exception as e:
logger.error(f"Error in get_device_detail_by_name: {e}")
raise DatabaseError(f"Error while fetching device detail: {e}")
def initialize_database():
Base.metadata.create_all(engine)