1
1
from __future__ import annotations
2
- import os .path
2
+
3
+ from pathlib import Path
4
+
3
5
import orjson as json
4
6
from cryptography .fernet import Fernet , InvalidToken
5
7
@@ -15,7 +17,13 @@ class PantherDB:
15
17
__return_dict : bool
16
18
__content : dict
17
19
18
- def __init__ (self , db_name : str | None = None , return_dict : bool = False , secret_key : bytes | None = None ):
20
+ def __init__ (
21
+ self ,
22
+ * ,
23
+ db_name : str | None = None ,
24
+ return_dict : bool = False ,
25
+ secret_key : bytes | None = None ,
26
+ ):
19
27
self .__return_dict = return_dict
20
28
self .__secret_key = secret_key
21
29
self .__fernet = Fernet (self .__secret_key ) if self .__secret_key else None
@@ -25,8 +33,7 @@ def __init__(self, db_name: str | None = None, return_dict: bool = False, secret
25
33
if not db_name .endswith ('pdb' ):
26
34
db_name = f'{ db_name } .pdb'
27
35
self .db_name = db_name
28
- if not os .path .exists (self .db_name ):
29
- open (self .db_name , 'w' ).close ()
36
+ Path (self .db_name ).touch (exist_ok = True )
30
37
31
38
def __str__ (self ) -> str :
32
39
self ._refresh ()
@@ -79,7 +86,7 @@ def collection(self, collection_name: str) -> PantherCollection:
79
86
db_name = self .db_name ,
80
87
collection_name = collection_name ,
81
88
return_dict = self .return_dict ,
82
- secret_key = self .secret_key
89
+ secret_key = self .secret_key ,
83
90
)
84
91
85
92
def close (self ):
@@ -89,16 +96,21 @@ def close(self):
89
96
class PantherCollection (PantherDB ):
90
97
__collection_name : str
91
98
92
- def __init__ (self , db_name : str , collection_name : str , return_dict : bool , secret_key : bytes ):
99
+ def __init__ (
100
+ self ,
101
+ * ,
102
+ db_name : str ,
103
+ collection_name : str ,
104
+ return_dict : bool ,
105
+ secret_key : bytes ,
106
+ ):
93
107
super ().__init__ (db_name = db_name , return_dict = return_dict , secret_key = secret_key )
94
108
self .__collection_name = collection_name
95
109
96
110
def __str__ (self ) -> str :
97
111
self ._refresh ()
98
112
99
- if self .collection_name not in self .content :
100
- result = ''
101
- elif documents := self .content [self .collection_name ]:
113
+ if self .collection_name not in self .content or (documents := self .content [self .collection_name ]):
102
114
result = ''
103
115
else :
104
116
result = '\n ' .join (f' { k } : { type (v ).__name__ } ' for k , v in documents [0 ].items ())
@@ -122,7 +134,7 @@ def __create_result(self, data: dict, /) -> PantherDocument | dict:
122
134
collection_name = self .collection_name ,
123
135
return_dict = self .return_dict ,
124
136
secret_key = self .secret_key ,
125
- ** data
137
+ ** data ,
126
138
)
127
139
128
140
def _write_collection (self , documents : list ) -> None :
@@ -136,7 +148,7 @@ def _drop_collection(self) -> None:
136
148
self ._write ()
137
149
138
150
def _get_collection (self ) -> list [dict ]:
139
- """return documents"""
151
+ """Return documents"""
140
152
self ._refresh ()
141
153
return self .content .get (self .collection_name , [])
142
154
@@ -185,7 +197,7 @@ def delete(self) -> None:
185
197
self .__check_is_panther_document ()
186
198
documents = self ._get_collection ()
187
199
for d in documents :
188
- if d .get ('_id' ) == self ._id : # NOQA : Unresolved References
200
+ if d .get ('_id' ) == self ._id : # noqa : Unresolved References
189
201
documents .remove (d )
190
202
self ._write_collection (documents )
191
203
break
@@ -246,7 +258,7 @@ def update(self, **kwargs) -> None:
246
258
self .__check_is_panther_document ()
247
259
documents = self ._get_collection ()
248
260
for d in documents :
249
- if d .get ('_id' ) == self ._id : # NOQA : Unresolved References
261
+ if d .get ('_id' ) == self ._id : # noqa : Unresolved References
250
262
for k , v in kwargs .items ():
251
263
d [k ] = v
252
264
setattr (self , k , v )
@@ -312,13 +324,21 @@ def drop(self) -> None:
312
324
class PantherDocument (PantherCollection ):
313
325
__data : dict
314
326
315
- def __init__ (self , db_name : str , collection_name : str , return_dict : bool , secret_key : bytes , ** kwargs ):
327
+ def __init__ (
328
+ self ,
329
+ * ,
330
+ db_name : str ,
331
+ collection_name : str ,
332
+ return_dict : bool ,
333
+ secret_key : bytes ,
334
+ ** kwargs ,
335
+ ):
316
336
self .__data = kwargs
317
337
super ().__init__ (
318
338
db_name = db_name ,
319
339
collection_name = collection_name ,
320
340
return_dict = return_dict ,
321
- secret_key = secret_key
341
+ secret_key = secret_key ,
322
342
)
323
343
324
344
def __str__ (self ) -> str :
@@ -341,7 +361,7 @@ def __setattr__(self, key, value):
341
361
'_PantherDB__content' ,
342
362
'_PantherDB__fernet' ,
343
363
'_PantherCollection__collection_name' ,
344
- '_PantherDocument__data'
364
+ '_PantherDocument__data' ,
345
365
]:
346
366
try :
347
367
object .__getattribute__ (self , key )
0 commit comments