ebook_homebrew.rdb   A
last analyzed

Complexity

Total Complexity 15

Size/Duplication

Total Lines 192
Duplicated Lines 0 %

Test Coverage

Coverage 91.14%

Importance

Changes 0
Metric Value
wmc 15
eloc 105
dl 0
loc 192
ccs 72
cts 79
cp 0.9114
rs 10
c 0
b 0
f 0

6 Methods

Rating   Name   Duplication   Size   Complexity  
A UploadedFile.__init__() 0 16 2
A UploadedFile.update_uploaded_file_last_index() 0 23 2
A UploadedFile.add_uploaded_file() 0 33 3
A UploadedFile.delete_uploaded_file() 0 25 2
A UploadedFile.get_all_uploaded_file() 0 31 3
A UploadedFile.get_uploaded_file() 0 37 3
1
# -*- coding: utf-8 -*-
2 1
"""Provides RDB execute
3
"""
4 1
import os
5
6 1
from sqlalchemy import create_engine
7 1
from sqlalchemy.orm import sessionmaker
8 1
from sqlalchemy.exc import SQLAlchemyError
9 1
from sqlalchemy.orm.exc import NoResultFound
10
11 1
from .models.uploaded_files_models import Base, UploadedFilesModel
12
13 1
from .utils.logging import get_logger
14
15 1
_logger = get_logger("rdb")
16
17
18 1
class UploadedFile:
19
    """Provides UploadFile Sqlite3 operation
20
    """
21
22 1
    def __init__(self, dbname="ebook-homebrew.sqlite3", echo_log=True):
23
        """Constructor
24
        Create Sqlite3 db file and session.
25
        Args:
26
            dbname (str): Sqlite3 db name, default is ebook-homebrew.sqlite3
27
            echo_log (bool): If True, echo DB queries.
28
        """
29 1
        self.dbname = dbname
30 1
        self.engine = create_engine(
31
            "sqlite:///{dbname}".format(dbname=self.dbname), echo=echo_log
32
        )
33 1
        Base.metadata.create_all(self.engine)
34 1
        if not os.path.isfile(self.dbname):
35
            Base.metadata.create_all(self.engine)
36 1
        Session = sessionmaker(bind=self.engine)
37 1
        self.session = Session()
38
39 1
    def add_uploaded_file(self, name, file_path, file_type, last_index):
40
        """Insert upload file
41
42
        Args:
43
            name (str): filename
44
            file_path (str): file path means upload ID
45
            file_type (str): ContentType like Image/png
46
            last_index (int): file index
47
48
        Returns:
49
            None: If Success
50
        Raises:
51
            SQLAlchemyError: SQL's Error
52
        """
53 1
        try:
54 1
            upload_files = UploadedFilesModel(
55
                name=name,
56
                file_path=file_path,
57
                file_type=file_type,
58
                last_index=last_index,
59
            )
60 1
            self.session.add(upload_files)
61 1
            self.session.commit()
62 1
        except SQLAlchemyError as err:
63 1
            _logger.exception(err)
64 1
            self.session.rollback()
65 1
            raise err
66 1
        except Exception as err:
67 1
            _logger.exception(err)
68 1
            self.session.rollback()
69 1
            raise err
70
        finally:
71 1
            self.session.close()
72
73 1
    def update_uploaded_file_last_index(self, table_id, last_index):
74
        """Update file last index.
75
76
        Args:
77
            table_id (str): id
78
            last_index (int): file last index
79
80
        Returns:
81
            None: If Success
82
        Raises:
83
            SQLAlchemyError: SQL's Error
84
        """
85 1
        try:
86 1
            query = self.session.query(UploadedFilesModel).with_lockmode("update")
87 1
            upload_file = query.filter(UploadedFilesModel.id == table_id).first()
88 1
            upload_file.last_index = last_index
89 1
            self.session.commit()
90 1
        except Exception as err:
91 1
            _logger.exception(err)
92 1
            self.session.rollback()
93 1
            raise err
94
        finally:
95 1
            self.session.close()
96
97 1
    def delete_uploaded_file(self, table_id):
98
        """Delete uploaded file with set id
99
100
        Args:
101
            table_id (str): id
102
103
        Returns:
104
            None: If Success
105
        Raises:
106
            SQLAlchemyError: SQL's Error
107
        """
108 1
        try:
109 1
            upload_file = (
110
                self.session.query(UploadedFilesModel)
111
                .filter(UploadedFilesModel.id == table_id)
112
                .first()
113
            )
114 1
            self.session.delete(upload_file)
115 1
            self.session.commit()
116 1
        except Exception as err:
117 1
            _logger.exception(err)
118 1
            self.session.rollback()
119 1
            raise err
120
        finally:
121 1
            self.session.close()
122
123 1
    def get_all_uploaded_file(self):
124
        """Get All uploaded files
125
126
        Returns:
127
            List[dict[int, str, str, str, int, datetime, datetime]]: uploaded files list
128
        Raises:
129
            SQLAlchemyError: SQL's Error
130
        """
131 1
        uploaded_file_list = []
132 1
        try:
133 1
            upload_files = self.session.query(UploadedFilesModel).all()
134 1
            for upload_file in upload_files:
135 1
                created_at = upload_file.created_at
136 1
                updated_at = upload_file.updated_at
137 1
                uploaded_file_list.append(
138
                    {
139
                        "id": upload_file.id,
140
                        "name": upload_file.name,
141
                        "file_path": upload_file.file_path,
142
                        "file_type": upload_file.file_type,
143
                        "last_index": upload_file.last_index,
144
                        "created_at": created_at.strftime("%Y-%m-%d %H:%M:%S"),
145
                        "updated_at": updated_at.strftime("%Y-%m-%d %H:%M:%S"),
146
                    }
147
                )
148 1
            return uploaded_file_list
149
        except Exception as err:
150
            _logger.exception(err)
151
            raise err
152
        finally:
153 1
            self.session.close()
154
155 1
    def get_uploaded_file(self, table_id):
156
        """Get uploaded file
157
158
        Args:
159
            table_id (str): id
160
161
        Returns:
162
            dict[int, str, str, str, int, datetime, datetime]: uploaded file
163
        Raises:
164
            SQLAlchemyError: SQL's Error
165
        """
166 1
        try:
167 1
            upload_file = (
168
                self.session.query(UploadedFilesModel)
169
                .filter(UploadedFilesModel.id == table_id)
170
                .first()
171
            )
172 1
            if upload_file:
173 1
                created_at = upload_file.created_at
174 1
                updated_at = upload_file.updated_at
175 1
                upload_file_dict = {
176
                    "id": upload_file.id,
177
                    "name": upload_file.name,
178
                    "file_path": upload_file.file_path,
179
                    "file_type": upload_file.file_type,
180
                    "last_index": upload_file.last_index,
181
                    "created_at": created_at.strftime("%Y-%m-%d %H:%M:%S"),
182
                    "updated_at": updated_at.strftime("%Y-%m-%d %H:%M:%S"),
183
                }
184
            else:
185 1
                upload_file_dict = {}
186 1
            return upload_file_dict
187
        except Exception as err:
188
            _logger.exception(err)
189
            raise err
190
        finally:
191
            self.session.close()
192