Issues in db_tools.py - New Issues - Inspection of "merge" - blabla1337/skf-flask - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 6bbd23...9f8f82 )

by Glenn

created 2018-08-20 12:34 UTC

skf/db_tools.py (2 issues)

Labels

Severity

Major 2

import os
from flask import Flask
from skf import settings
from shutil import copyfile
from skf.database import db
from skf.database.kb_items import kb_items
from sqlite3 import dbapi2 as sqlite3


app = Flask(__name__)

def connect_db():
    """Connects to the specific database."""
    rv = sqlite3.connect(os.path.join(app.root_path, settings.DATABASE))
    rv.row_factory = sqlite3.Row
    return rv


def init_db():
    """Initializes the database."""
    try:
        os.remove(os.path.join(app.root_path, settings.DATABASE))
        open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a')
        os.remove(os.path.join(app.root_path, 'db.sqlite_schema'))
        copyfile(os.path.join(app.root_path, "schema.sql"), os.path.join(app.root_path, 'db.sqlite_schema'))
        init_md_checklists()
        init_md_code_examples()
        init_md_knowledge_base()
        db = connect_db()
param = 5

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        with app.open_resource(os.path.join(app.root_path, 'db.sqlite_schema'), mode='r') as f:
            db.cursor().executescript(f.read())
        db.commit()
        return True
    except:
        return False


def update_db():
    """Update the database."""
    try:
        os.remove(os.path.join(app.root_path, 'db.sqlite_schema'))
        db = connect_db()
param = 5

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
        with app.open_resource(os.path.join(app.root_path, 'db.clean'), mode='r') as f:
            db.cursor().executescript(f.read())
        db.commit()
        init_md_checklists()
        init_md_code_examples()
        init_md_knowledge_base()
        with app.open_resource(os.path.join(app.root_path, 'db.sqlite_schema'), mode='r') as f:
            db.cursor().executescript(f.read())
        db.commit()
        return True
    except:
        return False


def get_db():
    """Opens a new database connection if there is none yet for the current application context."""
    if not hasattr(g, settings.DATABASE):
        g.sqlite_db = connect_db()
    return g.sqlite_db


def init_md_knowledge_base():
    """Converts markdown knowledge-base items to DB."""
    kb_dir = os.path.join(app.root_path, 'markdown/knowledge_base')
    try:
        for filename in os.listdir(kb_dir):
            if filename.endswith(".md"):
                name_raw = filename.split("-")
                kbID = name_raw[0].replace("_", " ")
                title = name_raw[3].replace("_", " ")
                file = os.path.join(kb_dir, filename)
                data = open(file, 'r')
                file_content = data.read()
                data.close()
                content_escaped = file_content.translate(str.maketrans({"'":  r"''", "-":  r"", "#":  r""}))
                query = "INSERT OR REPLACE INTO kb_items (kbID, content, title) VALUES ('"+kbID+"','"+content_escaped+"', '"+title+"'); \n"
                with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
                        myfile.write(query)
        print('Initialized the markdown knowledge-base.')
        return True
    except:
        return False


def init_md_code_examples():
    """Converts markdown code-example items to DB."""
    kb_dir = os.path.join(app.root_path, 'markdown/code_examples/')
    code_langs = ['asp', 'java', 'php', 'flask', 'django', 'go', 'ruby']
    try:
        for lang in code_langs:
            for filename in os.listdir(kb_dir+lang):
                if filename.endswith(".md"):
                    name_raw = filename.split("-")
                    title = name_raw[3].replace("_", " ")
                    file = os.path.join(kb_dir+lang, filename)
                    data = open(file, 'r')
                    file_content = data.read()
                    data.close()
                    content_escaped = file_content.translate(str.maketrans({"'":  r"''", "-":  r"", "#":  r""}))
                    query = "INSERT OR REPLACE INTO code_items (content, title, code_lang) VALUES ('"+content_escaped+"', '"+title+"', '"+lang+"'); \n"
                    with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
                            myfile.write(query)
        print('Initialized the markdown code-example.')
        return True
    except:
        return False


def init_md_checklists():
    """Converts markdown checklists items to DB."""
    kb_dir = os.path.join(app.root_path, 'markdown/checklists/')
    try:
        checklists = ['asvs', 'custom', 'masvs', 'pcidss32']
        for checklist in checklists:
            if checklist == "asvs":
                for filename in os.listdir(kb_dir+checklist):
                    if filename.endswith(".md"):
                        name_raw = filename.split("-")
                        level = name_raw[4].replace("_", " ")
                        kbid_raw = name_raw[6].split(".")
                        kb_id = kbid_raw[0]
                        if level == "0":
                            # For the ASVS categories
                            file = os.path.join(kb_dir+checklist, filename)
                            data = open(file, 'r')
                            file_content = data.read()
                            data.close()
                            checklistID_raw = file_content.split(":")
                            checklistID = checklistID_raw[0]
                            checklistID = checklistID.lstrip('V')
                            checklistID = checklistID+".0"
                        else :
                            # For the ASVS items
                            file = os.path.join(kb_dir+checklist, filename)
                            data = open(file, 'r')
                            file_content = data.read()
                            data.close()
                            checklistID_raw = file_content.split(" ")
                            checklistID = checklistID_raw[0]
                        file = os.path.join(kb_dir+checklist, filename)
                        data = open(file, 'r')
                        file_content = data.read()
                        data.close()
                        content = file_content.split(' ', 1)[1]
                        content_escaped = content.translate(str.maketrans({"'":  r"''", "-":  r"", "#":  r""}))
                        query = "INSERT OR REPLACE INTO checklists (checklist_type, checklistID, content, level, kbID) VALUES (0, '"+checklistID+"', '"+content_escaped+"', '"+level+"', '"+kb_id+"'); \n"
                        with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
                            myfile.write(query)
            if checklist == 'masvs':
                for filename in os.listdir(kb_dir+checklist):
                    if filename.endswith(".md"):
                        name_raw = filename.split("-")
                        level = name_raw[4].replace("_", " ")
                        kbid_raw = name_raw[6].split(".")
                        kb_id = kbid_raw[0]
                        if level == "0":
                            # For the MASVS categories
                            file = os.path.join(kb_dir+checklist, filename)
                            data = open(file, 'r')
                            file_content = data.read()
                            data.close()
                            checklistID_raw = file_content.split(":")
                            checklistID = checklistID_raw[0]
                            checklistID = checklistID.lstrip('V')
                            checklistID = checklistID+".0"
                        else :
                            # For the MASVS items
                            file = os.path.join(kb_dir+checklist, filename)
                            data = open(file, 'r')
                            file_content = data.read()
                            data.close()
                            checklistID_raw = file_content.split(" ")
                            checklistID = checklistID_raw[0]
                        file = os.path.join(kb_dir+checklist, filename)
                        data = open(file, 'r')
                        file_content = data.read()
                        data.close()
                        content = file_content.split(' ', 1)[1]
                        content_escaped = content.translate(str.maketrans({"'":  r"''", "-":  r"", "#":  r""}))
                        query = "INSERT OR REPLACE INTO checklists (checklist_type, checklistID, content, level, kbID) VALUES (1, '"+checklistID+"', '"+content_escaped+"', '"+level+"', '"+kb_id+"'); \n"
                        with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
                                myfile.write(query)
            if checklist == "pcidss32":
                for filename in os.listdir(kb_dir+checklist):
                    if filename.endswith(".md"):
                        name_raw = filename.split("-")
                        level = name_raw[4].replace("_", " ")
                        kbid_raw = name_raw[6].split(".")
                        kb_id = kbid_raw[0]
                        if level == "0":
                            # For the pcidss32 categories
                            file = os.path.join(kb_dir+checklist, filename)
                            data = open(file, 'r')
                            file_content = data.read()
                            data.close()
                            checklistID_raw = file_content.split(":")
                            checklistID = checklistID_raw[0]
                            checklistID = checklistID.lstrip('V')
                            checklistID = checklistID+".0"
                        else :
                            # For the pcidss32 items
                            file = os.path.join(kb_dir+checklist, filename)
                            data = open(file, 'r')
                            file_content = data.read()
                            data.close()
                            checklistID_raw = file_content.split(" ")
                            checklistID = checklistID_raw[0]
                        file = os.path.join(kb_dir+checklist, filename)
                        data = open(file, 'r')
                        file_content = data.read()
                        data.close()
                        content = file_content.split(' ', 1)[1]
                        content_escaped = content.translate(str.maketrans({"'":  r"''", "-":  r"", "#":  r""}))
                        query = "INSERT OR REPLACE INTO checklists (checklist_type, checklistID, content, level, kbID) VALUES (2, '"+checklistID+"', '"+content_escaped+"', '"+level+"', '"+kb_id+"'); \n"
                        with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
                            myfile.write(query)
            if checklist == "custom":
                for filename in os.listdir(kb_dir+checklist):
                    if filename.endswith(".md"):
                        name_raw = filename.split("-")
                        level = name_raw[4].replace("_", " ")
                        kbid_raw = name_raw[6].split(".")
                        kb_id = kbid_raw[0]
                        if level == "0":
                            # For the custom categories
                            file = os.path.join(kb_dir+checklist, filename)
                            data = open(file, 'r')
                            file_content = data.read()
                            data.close()
                            checklistID_raw = file_content.split(":")
                            checklistID = checklistID_raw[0]
                            checklistID = checklistID.lstrip('V')
                            checklistID = checklistID+".0"
                        else :
                            # For the custom items
                            file = os.path.join(kb_dir+checklist, filename)
                            data = open(file, 'r')
                            file_content = data.read()
                            data.close()
                            checklistID_raw = file_content.split(" ")
                            checklistID = checklistID_raw[0]
                        file = os.path.join(kb_dir+checklist, filename)
                        data = open(file, 'r')
                        file_content = data.read()
                        data.close()
                        content = file_content.split(' ', 1)[1]
                        content_escaped = content.translate(str.maketrans({"'":  r"''", "-":  r"", "#":  r""}))
                        query = "INSERT OR REPLACE INTO checklists (checklist_type, checklistID, content, level, kbID) VALUES (3, '"+checklistID+"', '"+content_escaped+"', '"+level+"', '"+kb_id+"'); \n"
                        with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
                            myfile.write(query)
        print('Initialized the markdown checklists.')
        return True
    except Exception as e:
        print('Exception in file db_tools, method init_md_checklists: ' + e)
        return False


1			import os
2			from flask import Flask
3			from skf import settings
4			from shutil import copyfile
5			from skf.database import db
6			from skf.database.kb_items import kb_items
7			from sqlite3 import dbapi2 as sqlite3
8
9
10			app = Flask(__name__)
11
12			def connect_db():
13			"""Connects to the specific database."""
14			rv = sqlite3.connect(os.path.join(app.root_path, settings.DATABASE))
15			rv.row_factory = sqlite3.Row
16			return rv
17
18
19			def init_db():
20			"""Initializes the database."""
21			try:
22			os.remove(os.path.join(app.root_path, settings.DATABASE))
23			open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a')
24			os.remove(os.path.join(app.root_path, 'db.sqlite_schema'))
25			copyfile(os.path.join(app.root_path, "schema.sql"), os.path.join(app.root_path, 'db.sqlite_schema'))
26			init_md_checklists()
27			init_md_code_examples()
28			init_md_knowledge_base()
29			db = connect_db()
			0 ignored issues – show Comprehensibility Bug introduced 2018-08-20 12:34 UTC by Report Bug Copy Issue Report Show Similar Issues like this `db` is re-defining a name which is already available in the outer-scope (previously defined on line `5`). It is generally a bad practice to shadow variables from the outer-scope. In most cases, this is done unintentionally and might lead to unexpected behavior: param = 5 class Foo: def __init__(self, param): # "param" would be flagged here self.param = param Loading history...
30			with app.open_resource(os.path.join(app.root_path, 'db.sqlite_schema'), mode='r') as f:
31			db.cursor().executescript(f.read())
32			db.commit()
33			return True
34			except:
35			return False
36
37
38			def update_db():
39			"""Update the database."""
40			try:
41			os.remove(os.path.join(app.root_path, 'db.sqlite_schema'))
42			db = connect_db()
			0 ignored issues – show Comprehensibility Bug introduced 2018-08-20 12:34 UTC by Report Bug Copy Issue Report Show Similar Issues like this `db` is re-defining a name which is already available in the outer-scope (previously defined on line `5`). It is generally a bad practice to shadow variables from the outer-scope. In most cases, this is done unintentionally and might lead to unexpected behavior: param = 5 class Foo: def __init__(self, param): # "param" would be flagged here self.param = param Loading history...
43			with app.open_resource(os.path.join(app.root_path, 'db.clean'), mode='r') as f:
44			db.cursor().executescript(f.read())
45			db.commit()
46			init_md_checklists()
47			init_md_code_examples()
48			init_md_knowledge_base()
49			with app.open_resource(os.path.join(app.root_path, 'db.sqlite_schema'), mode='r') as f:
50			db.cursor().executescript(f.read())
51			db.commit()
52			return True
53			except:
54			return False
55
56
57			def get_db():
58			"""Opens a new database connection if there is none yet for the current application context."""
59			if not hasattr(g, settings.DATABASE):
60			g.sqlite_db = connect_db()
61			return g.sqlite_db
62
63
64			def init_md_knowledge_base():
65			"""Converts markdown knowledge-base items to DB."""
66			kb_dir = os.path.join(app.root_path, 'markdown/knowledge_base')
67			try:
68			for filename in os.listdir(kb_dir):
69			if filename.endswith(".md"):
70			name_raw = filename.split("-")
71			kbID = name_raw[0].replace("_", " ")
72			title = name_raw[3].replace("_", " ")
73			file = os.path.join(kb_dir, filename)
74			data = open(file, 'r')
75			file_content = data.read()
76			data.close()
77			content_escaped = file_content.translate(str.maketrans({"'": r"''", "-": r"", "#": r""}))
78			query = "INSERT OR REPLACE INTO kb_items (kbID, content, title) VALUES ('"+kbID+"','"+content_escaped+"', '"+title+"'); \n"
79			with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
80			myfile.write(query)
81			print('Initialized the markdown knowledge-base.')
82			return True
83			except:
84			return False
85
86
87			def init_md_code_examples():
88			"""Converts markdown code-example items to DB."""
89			kb_dir = os.path.join(app.root_path, 'markdown/code_examples/')
90			code_langs = ['asp', 'java', 'php', 'flask', 'django', 'go', 'ruby']
91			try:
92			for lang in code_langs:
93			for filename in os.listdir(kb_dir+lang):
94			if filename.endswith(".md"):
95			name_raw = filename.split("-")
96			title = name_raw[3].replace("_", " ")
97			file = os.path.join(kb_dir+lang, filename)
98			data = open(file, 'r')
99			file_content = data.read()
100			data.close()
101			content_escaped = file_content.translate(str.maketrans({"'": r"''", "-": r"", "#": r""}))
102			query = "INSERT OR REPLACE INTO code_items (content, title, code_lang) VALUES ('"+content_escaped+"', '"+title+"', '"+lang+"'); \n"
103			with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
104			myfile.write(query)
105			print('Initialized the markdown code-example.')
106			return True
107			except:
108			return False
109
110
111			def init_md_checklists():
112			"""Converts markdown checklists items to DB."""
113			kb_dir = os.path.join(app.root_path, 'markdown/checklists/')
114			try:
115			checklists = ['asvs', 'custom', 'masvs', 'pcidss32']
116			for checklist in checklists:
117			if checklist == "asvs":
118			for filename in os.listdir(kb_dir+checklist):
119			if filename.endswith(".md"):
120			name_raw = filename.split("-")
121			level = name_raw[4].replace("_", " ")
122			kbid_raw = name_raw[6].split(".")
123			kb_id = kbid_raw[0]
124			if level == "0":
125			# For the ASVS categories
126			file = os.path.join(kb_dir+checklist, filename)
127			data = open(file, 'r')
128			file_content = data.read()
129			data.close()
130			checklistID_raw = file_content.split(":")
131			checklistID = checklistID_raw[0]
132			checklistID = checklistID.lstrip('V')
133			checklistID = checklistID+".0"
134			else :
135			# For the ASVS items
136			file = os.path.join(kb_dir+checklist, filename)
137			data = open(file, 'r')
138			file_content = data.read()
139			data.close()
140			checklistID_raw = file_content.split(" ")
141			checklistID = checklistID_raw[0]
142			file = os.path.join(kb_dir+checklist, filename)
143			data = open(file, 'r')
144			file_content = data.read()
145			data.close()
146			content = file_content.split(' ', 1)[1]
147			content_escaped = content.translate(str.maketrans({"'": r"''", "-": r"", "#": r""}))
148			query = "INSERT OR REPLACE INTO checklists (checklist_type, checklistID, content, level, kbID) VALUES (0, '"+checklistID+"', '"+content_escaped+"', '"+level+"', '"+kb_id+"'); \n"
149			with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
150			myfile.write(query)
151			if checklist == 'masvs':
152			for filename in os.listdir(kb_dir+checklist):
153			if filename.endswith(".md"):
154			name_raw = filename.split("-")
155			level = name_raw[4].replace("_", " ")
156			kbid_raw = name_raw[6].split(".")
157			kb_id = kbid_raw[0]
158			if level == "0":
159			# For the MASVS categories
160			file = os.path.join(kb_dir+checklist, filename)
161			data = open(file, 'r')
162			file_content = data.read()
163			data.close()
164			checklistID_raw = file_content.split(":")
165			checklistID = checklistID_raw[0]
166			checklistID = checklistID.lstrip('V')
167			checklistID = checklistID+".0"
168			else :
169			# For the MASVS items
170			file = os.path.join(kb_dir+checklist, filename)
171			data = open(file, 'r')
172			file_content = data.read()
173			data.close()
174			checklistID_raw = file_content.split(" ")
175			checklistID = checklistID_raw[0]
176			file = os.path.join(kb_dir+checklist, filename)
177			data = open(file, 'r')
178			file_content = data.read()
179			data.close()
180			content = file_content.split(' ', 1)[1]
181			content_escaped = content.translate(str.maketrans({"'": r"''", "-": r"", "#": r""}))
182			query = "INSERT OR REPLACE INTO checklists (checklist_type, checklistID, content, level, kbID) VALUES (1, '"+checklistID+"', '"+content_escaped+"', '"+level+"', '"+kb_id+"'); \n"
183			with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
184			myfile.write(query)
185			if checklist == "pcidss32":
186			for filename in os.listdir(kb_dir+checklist):
187			if filename.endswith(".md"):
188			name_raw = filename.split("-")
189			level = name_raw[4].replace("_", " ")
190			kbid_raw = name_raw[6].split(".")
191			kb_id = kbid_raw[0]
192			if level == "0":
193			# For the pcidss32 categories
194			file = os.path.join(kb_dir+checklist, filename)
195			data = open(file, 'r')
196			file_content = data.read()
197			data.close()
198			checklistID_raw = file_content.split(":")
199			checklistID = checklistID_raw[0]
200			checklistID = checklistID.lstrip('V')
201			checklistID = checklistID+".0"
202			else :
203			# For the pcidss32 items
204			file = os.path.join(kb_dir+checklist, filename)
205			data = open(file, 'r')
206			file_content = data.read()
207			data.close()
208			checklistID_raw = file_content.split(" ")
209			checklistID = checklistID_raw[0]
210			file = os.path.join(kb_dir+checklist, filename)
211			data = open(file, 'r')
212			file_content = data.read()
213			data.close()
214			content = file_content.split(' ', 1)[1]
215			content_escaped = content.translate(str.maketrans({"'": r"''", "-": r"", "#": r""}))
216			query = "INSERT OR REPLACE INTO checklists (checklist_type, checklistID, content, level, kbID) VALUES (2, '"+checklistID+"', '"+content_escaped+"', '"+level+"', '"+kb_id+"'); \n"
217			with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
218			myfile.write(query)
219			if checklist == "custom":
220			for filename in os.listdir(kb_dir+checklist):
221			if filename.endswith(".md"):
222			name_raw = filename.split("-")
223			level = name_raw[4].replace("_", " ")
224			kbid_raw = name_raw[6].split(".")
225			kb_id = kbid_raw[0]
226			if level == "0":
227			# For the custom categories
228			file = os.path.join(kb_dir+checklist, filename)
229			data = open(file, 'r')
230			file_content = data.read()
231			data.close()
232			checklistID_raw = file_content.split(":")
233			checklistID = checklistID_raw[0]
234			checklistID = checklistID.lstrip('V')
235			checklistID = checklistID+".0"
236			else :
237			# For the custom items
238			file = os.path.join(kb_dir+checklist, filename)
239			data = open(file, 'r')
240			file_content = data.read()
241			data.close()
242			checklistID_raw = file_content.split(" ")
243			checklistID = checklistID_raw[0]
244			file = os.path.join(kb_dir+checklist, filename)
245			data = open(file, 'r')
246			file_content = data.read()
247			data.close()
248			content = file_content.split(' ', 1)[1]
249			content_escaped = content.translate(str.maketrans({"'": r"''", "-": r"", "#": r""}))
250			query = "INSERT OR REPLACE INTO checklists (checklist_type, checklistID, content, level, kbID) VALUES (3, '"+checklistID+"', '"+content_escaped+"', '"+level+"', '"+kb_id+"'); \n"
251			with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
252			myfile.write(query)
253			print('Initialized the markdown checklists.')
254			return True
255			except Exception as e:
256			print('Exception in file db_tools, method init_md_checklists: ' + e)
257			return False
258

blabla1337 / skf-flask

Push — master ( 6bbd23...9f8f82 )

skf/db_tools.py (2 issues)

Labels

Severity

Introduced By

Duplication Side-by-Side

Filter issues like