update_db() - Code Metrics - Inspection of "MASVS initial commit, ASVS kbID correction, closes..." - blabla1337/skf-flask - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#440)

unknown

created 2018-02-27 11:28 UTC

update_db() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	22
rs	8.9197
cc	4

import os
from skf import settings
from shutil import copyfile
from flask import Flask
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:
        if (os.path.exists(os.path.join(app.root_path, settings.DATABASE))):
            os.remove(os.path.join(app.root_path, settings.DATABASE))
        open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a')
        if (os.path.exists(os.path.join(app.root_path, 'db.sqlite_schema'))):
            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()
        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 Exception as e:
        print('Exception in file db_tools, method init_db: ' + e)
        return False


def update_db():
    """Update the database."""
    try:
        if (os.path.exists(os.path.join(app.root_path, 'db.sqlite_schema'))):
            os.remove(os.path.join(app.root_path, 'db.sqlite_schema'))
        db = connect_db()
        db.execute("DELETE FROM kb_items")
        db.execute("DELETE FROM code_items")
        db.execute("DELETE FROM checklists")
        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 Exception as e:
        print('Exception in file db_tools, method update_db: ' + e)
        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 Exception as e:
        print('Exception in file db_tools, method init_md_knowledge_base: ' + e)
        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 Exception as e:
        print('Exception in file db_tools, method init_md_code_examples: ' + e)
        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', 'pcidss', 'custom']
        checklists = ['asvs', 'custom', 'masvs']
        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 (checklistID, content, level, kbID) VALUES ('"+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 (checklistID, content, level, kbID) VALUES ('"+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 skf import settings
3			from shutil import copyfile
4			from flask import Flask
5			from sqlite3 import dbapi2 as sqlite3
6
7
8			app = Flask(__name__)
9
10			def connect_db():
11			"""Connects to the specific database."""
12			rv = sqlite3.connect(os.path.join(app.root_path, settings.DATABASE))
13			rv.row_factory = sqlite3.Row
14			return rv
15
16
17			def init_db():
18			"""Initializes the database."""
19			try:
20			if (os.path.exists(os.path.join(app.root_path, settings.DATABASE))):
21			os.remove(os.path.join(app.root_path, settings.DATABASE))
22			open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a')
23			if (os.path.exists(os.path.join(app.root_path, 'db.sqlite_schema'))):
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()
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 Exception as e:
35			print('Exception in file db_tools, method init_db: ' + e)
36			return False
37
38
39			def update_db():
40			"""Update the database."""
41			try:
42			if (os.path.exists(os.path.join(app.root_path, 'db.sqlite_schema'))):
43			os.remove(os.path.join(app.root_path, 'db.sqlite_schema'))
44			db = connect_db()
45			db.execute("DELETE FROM kb_items")
46			db.execute("DELETE FROM code_items")
47			db.execute("DELETE FROM checklists")
48			db.commit()
49
50			init_md_checklists()
51			init_md_code_examples()
52			init_md_knowledge_base()
53
54			with app.open_resource(os.path.join(app.root_path, 'db.sqlite_schema'), mode='r') as f:
55			db.cursor().executescript(f.read())
56			db.commit()
57			return True
58			except Exception as e:
59			print('Exception in file db_tools, method update_db: ' + e)
60			return False
61
62
63			def get_db():
64			"""Opens a new database connection if there is none yet for the current application context."""
65			if not hasattr(g, settings.DATABASE):
66			g.sqlite_db = connect_db()
67			return g.sqlite_db
68
69
70			def init_md_knowledge_base():
71			"""Converts markdown knowledge-base items to DB."""
72			kb_dir = os.path.join(app.root_path, 'markdown/knowledge_base')
73			try:
74			for filename in os.listdir(kb_dir):
75			if filename.endswith(".md"):
76			name_raw = filename.split("-")
77			kbID = name_raw[0].replace("_", " ")
78			title = name_raw[3].replace("_", " ")
79			file = os.path.join(kb_dir, filename)
80			data = open(file, 'r')
81			file_content = data.read()
82			data.close()
83			content_escaped = file_content.translate(str.maketrans({"'": r"''", "-": r"", "#": r""}))
84			query = "INSERT OR REPLACE INTO kb_items (kbID, content, title) VALUES ('"+kbID+"','"+content_escaped+"', '"+title+"'); \n"
85			with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
86			myfile.write(query)
87			print('Initialized the markdown knowledge-base.')
88			return True
89			except Exception as e:
90			print('Exception in file db_tools, method init_md_knowledge_base: ' + e)
91			return False
92
93
94			def init_md_code_examples():
95			"""Converts markdown code-example items to DB."""
96			kb_dir = os.path.join(app.root_path, 'markdown/code_examples/')
97			code_langs = ['asp', 'java', 'php', 'flask', 'django', 'go', 'ruby']
98			try:
99			for lang in code_langs:
100			for filename in os.listdir(kb_dir+lang):
101			if filename.endswith(".md"):
102			name_raw = filename.split("-")
103			title = name_raw[3].replace("_", " ")
104			file = os.path.join(kb_dir+lang, filename)
105			data = open(file, 'r')
106			file_content = data.read()
107			data.close()
108			content_escaped = file_content.translate(str.maketrans({"'": r"''", "-": r"", "#": r""}))
109			query = "INSERT OR REPLACE INTO code_items (content, title, code_lang) VALUES ('"+content_escaped+"', '"+title+"', '"+lang+"'); \n"
110			with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
111			myfile.write(query)
112			print('Initialized the markdown code-example.')
113			return True
114			except Exception as e:
115			print('Exception in file db_tools, method init_md_code_examples: ' + e)
116			return False
117
118
119			def init_md_checklists():
120			"""Converts markdown checklists items to DB."""
121			kb_dir = os.path.join(app.root_path, 'markdown/checklists/')
122			try:
123			#checklists = ['asvs', 'pcidss', 'custom']
124			checklists = ['asvs', 'custom', 'masvs']
125			for checklist in checklists:
126			if checklist == "asvs":
127			for filename in os.listdir(kb_dir+checklist):
128			if filename.endswith(".md"):
129			name_raw = filename.split("-")
130			level = name_raw[4].replace("_", " ")
131			kbid_raw = name_raw[6].split(".")
132			kb_id = kbid_raw[0]
133			if level == "0":
134			# For the ASVS categories
135			file = os.path.join(kb_dir+checklist, filename)
136			data = open(file, 'r')
137			file_content = data.read()
138			data.close()
139			checklistID_raw = file_content.split(":")
140			checklistID = checklistID_raw[0]
141			checklistID = checklistID.lstrip('V')
142			checklistID = checklistID+".0"
143			else :
144			# For the ASVS items
145			file = os.path.join(kb_dir+checklist, filename)
146			data = open(file, 'r')
147			file_content = data.read()
148			data.close()
149			checklistID_raw = file_content.split(" ")
150			checklistID = checklistID_raw[0]
151			file = os.path.join(kb_dir+checklist, filename)
152			data = open(file, 'r')
153			file_content = data.read()
154			data.close()
155			content = file_content.split(' ', 1)[1]
156			content_escaped = content.translate(str.maketrans({"'": r"''", "-": r"", "#": r""}))
157			query = "INSERT OR REPLACE INTO checklists (checklistID, content, level, kbID) VALUES ('"+checklistID+"', '"+content_escaped+"', '"+level+"', '"+kb_id+"'); \n"
158			with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
159			myfile.write(query)
160			if checklist == 'masvs':
161			for filename in os.listdir(kb_dir+checklist):
162			if filename.endswith(".md"):
163			name_raw = filename.split("-")
164			level = name_raw[4].replace("_", " ")
165			kbid_raw = name_raw[6].split(".")
166			kb_id = kbid_raw[0]
167			if level == "0":
168			# For the MASVS categories
169			file = os.path.join(kb_dir+checklist, filename)
170			data = open(file, 'r')
171			file_content = data.read()
172			data.close()
173			checklistID_raw = file_content.split(":")
174			checklistID = checklistID_raw[0]
175			checklistID = checklistID.lstrip('V')
176			checklistID = checklistID+".0"
177			else :
178			# For the MASVS items
179			file = os.path.join(kb_dir+checklist, filename)
180			data = open(file, 'r')
181			file_content = data.read()
182			data.close()
183			checklistID_raw = file_content.split(" ")
184			checklistID = checklistID_raw[0]
185			file = os.path.join(kb_dir+checklist, filename)
186			data = open(file, 'r')
187			file_content = data.read()
188			data.close()
189			content = file_content.split(' ', 1)[1]
190			content_escaped = content.translate(str.maketrans({"'": r"''", "-": r"", "#": r""}))
191			query = "INSERT OR REPLACE INTO checklists (checklistID, content, level, kbID) VALUES ('"+checklistID+"', '"+content_escaped+"', '"+level+"', '"+kb_id+"'); \n"
192			with open(os.path.join(app.root_path, 'db.sqlite_schema'), 'a') as myfile:
193			myfile.write(query)
194			print('Initialized the markdown checklists.')
195			return True
196			except Exception as e:
197			print('Exception in file db_tools, method init_md_checklists: ' + e)
198			return False

blabla1337 / skf-flask

Pull Request — master (#440)

update_db() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like