1
1
"""
2
2
Storage Module
3
3
4
- This module provides a class for managing SQLite database operations specifically
4
+ Provides a class for managing SQLite database operations specifically
5
5
for Dynamic DNS (DDNS) services. The `Storage` class handles the creation,
6
6
updating, and retrieval of domain records in a SQLite database.
7
7
"""
8
+
8
9
import sqlite3
9
10
import logging
10
11
from typing import Optional , Callable , Any , Tuple
11
12
from datetime import datetime
12
13
14
+
13
15
class Storage :
14
16
"""
15
17
A class to manage SQLite database operations for DDNS services.
@@ -18,15 +20,18 @@ class Storage:
18
20
in a SQLite database.
19
21
"""
20
22
21
- def __init__ (self , filename : str = ' py_ddns.db' ):
23
+ def __init__ (self , filename : str = " py_ddns.db" ):
22
24
self .connection = sqlite3 .connect (filename )
23
25
self .cursor = self .connection .cursor ()
24
26
25
27
self .create_tables ()
26
28
27
29
@staticmethod
28
30
def handle_sqlite_error (func : Callable ) -> Callable :
29
- """ Decorator utilized to handle all errors invlovling the SQLite Database."""
31
+ """
32
+ Decorator utilized to handle all errors involving the SQLite Database.
33
+ """
34
+
30
35
def wrapper (* args , ** kwargs ) -> Any :
31
36
32
37
try :
@@ -45,11 +50,12 @@ def wrapper(*args, **kwargs) -> Any:
45
50
except sqlite3 .IntegrityError as err :
46
51
logging .error ("SQLite Integrity Error: %s" , err )
47
52
raise
53
+
48
54
return wrapper
49
55
50
56
@handle_sqlite_error
51
57
def create_tables (self ) -> None :
52
- """ Method to create all SQLite tables utilized by pyddns """
58
+ """Method to create all SQLite tables utilized by pyddns"""
53
59
54
60
sql = """
55
61
CREATE TABLE IF NOT EXISTS domains (
@@ -65,13 +71,17 @@ def create_tables(self) -> None:
65
71
"""
66
72
self .cursor .execute (sql )
67
73
self .connection .commit ()
68
- logging .debug ("SQLite: Successfully verified that the domains table is present." )
74
+ logging .debug (
75
+ "SQLite: Successfully verified that the domains table is present."
76
+ )
69
77
70
78
@handle_sqlite_error
71
79
def drop_tables (self ) -> None :
72
- """ Method to drop all SQLite tables utilized by pyddns """
80
+ """Method to drop all SQLite tables utilized by pyddns"""
73
81
74
- logging .warning ("one or more database tables have been requested to be dropped." )
82
+ logging .warning (
83
+ "one or more database tables have been requested to be dropped."
84
+ )
75
85
76
86
sql = """
77
87
DROP TABLE IF EXISTS domains;
@@ -83,39 +93,61 @@ def drop_tables(self) -> None:
83
93
84
94
@handle_sqlite_error
85
95
def add_service (
86
- self , service_name : str , domain_name : str , current_ip : str , record_id : Optional [str ] = None
87
- ) -> None :
88
- """ Adds a service to the SQLite database and sets a created_at timestamp """
96
+ self ,
97
+ service_name : str ,
98
+ domain_name : str ,
99
+ current_ip : str ,
100
+ record_id : Optional [str ] = None ,
101
+ ) -> None :
102
+ """
103
+ Adds a service to the SQLite database and sets a created_at timestamp
104
+ """
89
105
90
106
sql = """
91
107
INSERT INTO domains(service, domain_name, current_ip, record_id)
92
108
VALUES(?, ?, ?, ?)
93
109
"""
94
- self .cursor .execute (sql , (service_name , domain_name , current_ip , record_id ))
110
+ self .cursor .execute (
111
+ sql , (service_name , domain_name , current_ip , record_id )
112
+ )
95
113
self .connection .commit ()
96
114
logging .debug (
97
- "SQLite: Adding service: %s, Domain: %s, IP: %s" , service_name , domain_name , current_ip
115
+ "SQLite: Adding service: %s, Domain: %s, IP: %s" ,
116
+ service_name ,
117
+ domain_name ,
118
+ current_ip ,
98
119
)
99
120
logging .info ("SQLite: Sucessfully added %s to database." , service_name )
100
121
101
122
@handle_sqlite_error
102
- def update_ip (self , service_name : str , domain_name : str , current_ip : str ) -> None :
103
- """ Updated the domain name's IP address in the SQLite database. """
123
+ def update_ip (
124
+ self , service_name : str , domain_name : str , current_ip : str
125
+ ) -> None :
126
+ """Updated the domain name's IP address in the SQLite database."""
104
127
105
128
sql = """
106
129
UPDATE domains
107
130
SET service = COALESCE(?, service),
108
131
current_ip = COALESCE(?, current_ip),
109
132
last_updated = CURRENT_TIMESTAMP
110
- WHERE domain_name = ?
133
+ WHERE domain_name = ?
111
134
"""
112
135
self .cursor .execute (sql , (service_name , current_ip , domain_name ))
113
136
self .connection .commit ()
114
- logging .info ("SQLite: Updated %s on %s to %s" , domain_name , service_name , current_ip )
137
+ logging .info (
138
+ "SQLite: Updated %s on %s to %s" ,
139
+ domain_name ,
140
+ service_name ,
141
+ current_ip ,
142
+ )
115
143
116
144
@handle_sqlite_error
117
- def retrieve_record (self , domain_name : str ) -> Optional [Tuple [str , datetime , str ]]:
118
- """ Retrieves IP adress, last_updated, and record_id from SQLite databse if present. """
145
+ def retrieve_record (
146
+ self , domain_name : str
147
+ ) -> Optional [Tuple [str , datetime , str ]]:
148
+ """
149
+ Retrieves IP address, last_updated, and record_id from SQLite database
150
+ """
119
151
120
152
sql = """
121
153
SELECT current_ip, last_updated, record_id FROM domains
0 commit comments