build.resources.models.RequestModel.fetch_for_logged_in_user() - Code Metrics - Inspection of "#158114970 BUILT new endpoints & applied security..." - antokish/Maintenance-Tracker - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — develop (#32)

by Antony

created 2018-06-19 23:34 UTC

RequestModel.fetch_for_logged_in_user() A

↳ Parent: build.resources.models

Complexity

Conditions

Size

Total Lines	23
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	15
dl	0
loc	23
rs	9.0856
c	0
b	0
f	0
cc	3
nop	1

from flask import Flask
import datetime
import psycopg2

from passlib.hash import pbkdf2_sha256 as sha256
from run import create_app
from config import dbconfig, filename ,section, app_config, DATABASE_URL
from flask_bcrypt import Bcrypt

app = Flask(__name__)

def insert_to_db(self, username, password, firstname, lastname, role, created_on):
    """insert a new user into database"""

    query = """INSERT INTO tb_users (username, password, firstname, lastname,role, created_on)
                VALUES(%s,%s,%s,%s,%s,%s)"""
    
    conn = None
    user_id = None
    try:
        #params = dbconfig(filename, section)
        conn = psycopg2.connect(DATABASE_URL())

        cur = conn.cursor()
        cur.execute(query,(username, password, firstname, 
                    lastname,role, created_on,))

               
        conn.commit()
        user_id = cur.fetchone()[0]
        cur.close()
    except (Exception, psycopg2.DatabaseError) as error:
        return error
    finally:
        if conn is not None:
            conn.close()
    return user_id

def find_by_username(username):
    query = """SELECT user_id,username,password FROM tb_users WHERE username=(%s)"""
    
    conn = None
    result = None
    try:
        #params = dbconfig(filename, section)
        conn = psycopg2.connect(DATABASE_URL())

        cur = conn.cursor()
        cur.execute(query,(username,))

        result = cur.fetchone()
        
        #print(result)
        
        cur.close()

    except (Exception, psycopg2.DatabaseError) as error:
        print(error)
    finally:
        if conn is not None:
            conn.close()
    return result

def return_all():
    query = """select array_to_json(array_agg(row_to_json(t))) from (  
                SELECT * FROM tb_users) t"""
    
    conn = None
    result = None

    try:
        #params = dbconfig(filename, section)

        conn = psycopg2.connect(DATABASE_URL())

        cur = conn.cursor()
        cur.execute(query)

        result = cur.fetchall()

        cur.close()


    except (Exception, psycopg2.DatabaseError) as error:
        return str(error)
    finally:
        if conn is not None:
            conn.close()
    return result


def hash_password(password):
    #generate hashed string to store in db
    return sha256.hash(password)


def verify_hash(password, hash):
    #check the given pass
    return sha256.verify(password, hash)

def is_admin(username):
    #check if user is admin
    query = """SELECT username from tb_users WHERE username=(%s) AND role=True"""
    
    conn = None
    result = None
    try:
        conn = psycopg2.connect(DATABASE_URL())
        
        cur = conn.cursor()

        cur.execute(query, (username,))

        result = cur.fetchone()

        cur.close()

        if result is None or result == "":
            return False

    except (Exception, psycopg2.DatabaseError) as error:
        return "iko shida " + str(error)
    
    finally:
        if conn is not None:
            conn.close()

    return True

def get_id(username):
    pass

class RevokedTokenModel(object):
    pass

class RequestModel(object):
    def fetch_all_request():
        """get all request"""
        query = """select array_to_json(array_agg(row_to_json(t))) from (
                    select * from tb_request) t"""
        
        conn = None
        results = None
        try:
            conn=psycopg2.connect(DATABASE_URL())
            cur = conn.cursor()

            cur.execute(query)
            results = cur.fetchall()

            cur.close()

        except (Exception, psycopg2.DatabaseError) as error:
            return error
        finally:
            if conn is not None:
                conn.close()
        
        return results
    
    def fetch_for_logged_in_user(requestor):
        """get all request for logged in user"""
        query = """select array_to_json(array_agg(row_to_json(t))) from (
                    select * from tb_request WHERE requestor=(%s)) t"""
        
        conn = None
        results = None
        try:
            conn=psycopg2.connect(DATABASE_URL())
            cur = conn.cursor()

            cur.execute(query,(requestor,))
            results = cur.fetchall()

            cur.close()

        except (Exception, psycopg2.DatabaseError) as error:
            return error
        finally:
            if conn is not None:
                conn.close()
        
        return results
    
    def fetch_request_by_id(self, request_id):
        """get a request for a logged in user"""
        query = """select row_to_json(tb_request) from tb_request WHERE request_id=(%s)"""
        
        conn = None
        results = None
        try:
            conn=psycopg2.connect(DATABASE_URL())
            cur = conn.cursor()

            cur.execute(query,(request_id,))
            results = cur.fetchall()

            cur.close()

        except (Exception, psycopg2.DatabaseError) as error:
            return error
        finally:
            if conn is not None:
                conn.close()
        
        return results


    def create_request(self, requestor,request_type,status,description,created_on):
        """create a request"""
        sql = """INSERT into tb_request (requestor,request_type,status,description,created_on)
                VALUES(%s,%s,%s,%s,%s)"""
        conn = None

        try:
            conn = psycopg2.connect(DATABASE_URL())
            cur = conn.cursor()

            cur.execute(sql, (requestor,request_type,status,description,created_on,))

            conn.commit()
            cur.close()

        except (Exception, psycopg2.DatabaseError) as error:
            return error
        finally:
            if conn is not None:
                conn.close()
        return "Data saved succesfully"
    
    def modify_request(request_type,description,last_modified,request_id):
        """Edit or modify a request"""
        query = """UPDATE tb_request SET request_type=%s,description=%s,
                    last_modified=%s WHERE request_id=%s;"""
    
        conn = None
        user_id = None
        try:
            #params = dbconfig(filename, section)
            conn = psycopg2.connect(DATABASE_URL())

            cur = conn.cursor()
            cur.execute(query,(request_type,description,last_modified,request_id,))

                
            conn.commit()
            cur.close()
        except (Exception, psycopg2.DatabaseError) as error:
            return error
        finally:
            if conn is not None:
                conn.close()
        return user_id
    
    def request_action(self, status, last_modified, request_id):
        #PUT /request/<id>/approve
        sql = """UPDATE tb_request SET status=%s, last_modified=%s WHERE request_id=%s"""

        conn = None
        try:
            conn = psycopg2.connect(DATABASE_URL())
            cur = conn.cursor()
            cur.execute(sql,(status,last_modified, request_id,))

                
            conn.commit()
            cur.close()


        except (Exception, psycopg2.DatabaseError) as error:
            return error
        
        finally:
            if conn is not None:
                conn.close()
    

1			from flask import Flask
2			import datetime
3			import psycopg2
4
5			from passlib.hash import pbkdf2_sha256 as sha256
6			from run import create_app
7			from config import dbconfig, filename ,section, app_config, DATABASE_URL
8			from flask_bcrypt import Bcrypt
9
10			app = Flask(__name__)
11
12			def insert_to_db(self, username, password, firstname, lastname, role, created_on):
13			"""insert a new user into database"""
14
15			query = """INSERT INTO tb_users (username, password, firstname, lastname,role, created_on)
16			VALUES(%s,%s,%s,%s,%s,%s)"""
17
18			conn = None
19			user_id = None
20			try:
21			#params = dbconfig(filename, section)
22			conn = psycopg2.connect(DATABASE_URL())
23
24			cur = conn.cursor()
25			cur.execute(query,(username, password, firstname,
26			lastname,role, created_on,))
27
28
29			conn.commit()
30			user_id = cur.fetchone()[0]
31			cur.close()
32			except (Exception, psycopg2.DatabaseError) as error:
33			return error
34			finally:
35			if conn is not None:
36			conn.close()
37			return user_id
38
39			def find_by_username(username):
40			query = """SELECT user_id,username,password FROM tb_users WHERE username=(%s)"""
41
42			conn = None
43			result = None
44			try:
45			#params = dbconfig(filename, section)
46			conn = psycopg2.connect(DATABASE_URL())
47
48			cur = conn.cursor()
49			cur.execute(query,(username,))
50
51			result = cur.fetchone()
52
53			#print(result)
54
55			cur.close()
56
57			except (Exception, psycopg2.DatabaseError) as error:
58			print(error)
59			finally:
60			if conn is not None:
61			conn.close()
62			return result
63
64			def return_all():
65			query = """select array_to_json(array_agg(row_to_json(t))) from (
66			SELECT * FROM tb_users) t"""
67
68			conn = None
69			result = None
70
71			try:
72			#params = dbconfig(filename, section)
73
74			conn = psycopg2.connect(DATABASE_URL())
75
76			cur = conn.cursor()
77			cur.execute(query)
78
79			result = cur.fetchall()
80
81			cur.close()
82
83
84			except (Exception, psycopg2.DatabaseError) as error:
85			return str(error)
86			finally:
87			if conn is not None:
88			conn.close()
89			return result
90
91
92			def hash_password(password):
93			#generate hashed string to store in db
94			return sha256.hash(password)
95
96
97			def verify_hash(password, hash):
98			#check the given pass
99			return sha256.verify(password, hash)
100
101			def is_admin(username):
102			#check if user is admin
103			query = """SELECT username from tb_users WHERE username=(%s) AND role=True"""
104
105			conn = None
106			result = None
107			try:
108			conn = psycopg2.connect(DATABASE_URL())
109
110			cur = conn.cursor()
111
112			cur.execute(query, (username,))
113
114			result = cur.fetchone()
115
116			cur.close()
117
118			if result is None or result == "":
119			return False
120
121			except (Exception, psycopg2.DatabaseError) as error:
122			return "iko shida " + str(error)
123
124			finally:
125			if conn is not None:
126			conn.close()
127
128			return True
129
130			def get_id(username):
131			pass
132
133			class RevokedTokenModel(object):
134			pass
135
136			class RequestModel(object):
137			def fetch_all_request():
138			"""get all request"""
139			query = """select array_to_json(array_agg(row_to_json(t))) from (
140			select * from tb_request) t"""
141
142			conn = None
143			results = None
144			try:
145			conn=psycopg2.connect(DATABASE_URL())
146			cur = conn.cursor()
147
148			cur.execute(query)
149			results = cur.fetchall()
150
151			cur.close()
152
153			except (Exception, psycopg2.DatabaseError) as error:
154			return error
155			finally:
156			if conn is not None:
157			conn.close()
158
159			return results
160
161			def fetch_for_logged_in_user(requestor):
162			"""get all request for logged in user"""
163			query = """select array_to_json(array_agg(row_to_json(t))) from (
164			select * from tb_request WHERE requestor=(%s)) t"""
165
166			conn = None
167			results = None
168			try:
169			conn=psycopg2.connect(DATABASE_URL())
170			cur = conn.cursor()
171
172			cur.execute(query,(requestor,))
173			results = cur.fetchall()
174
175			cur.close()
176
177			except (Exception, psycopg2.DatabaseError) as error:
178			return error
179			finally:
180			if conn is not None:
181			conn.close()
182
183			return results
184
185			def fetch_request_by_id(self, request_id):
186			"""get a request for a logged in user"""
187			query = """select row_to_json(tb_request) from tb_request WHERE request_id=(%s)"""
188
189			conn = None
190			results = None
191			try:
192			conn=psycopg2.connect(DATABASE_URL())
193			cur = conn.cursor()
194
195			cur.execute(query,(request_id,))
196			results = cur.fetchall()
197
198			cur.close()
199
200			except (Exception, psycopg2.DatabaseError) as error:
201			return error
202			finally:
203			if conn is not None:
204			conn.close()
205
206			return results
207
208
209			def create_request(self, requestor,request_type,status,description,created_on):
210			"""create a request"""
211			sql = """INSERT into tb_request (requestor,request_type,status,description,created_on)
212			VALUES(%s,%s,%s,%s,%s)"""
213			conn = None
214
215			try:
216			conn = psycopg2.connect(DATABASE_URL())
217			cur = conn.cursor()
218
219			cur.execute(sql, (requestor,request_type,status,description,created_on,))
220
221			conn.commit()
222			cur.close()
223
224			except (Exception, psycopg2.DatabaseError) as error:
225			return error
226			finally:
227			if conn is not None:
228			conn.close()
229			return "Data saved succesfully"
230
231			def modify_request(request_type,description,last_modified,request_id):
232			"""Edit or modify a request"""
233			query = """UPDATE tb_request SET request_type=%s,description=%s,
234			last_modified=%s WHERE request_id=%s;"""
235
236			conn = None
237			user_id = None
238			try:
239			#params = dbconfig(filename, section)
240			conn = psycopg2.connect(DATABASE_URL())
241
242			cur = conn.cursor()
243			cur.execute(query,(request_type,description,last_modified,request_id,))
244
245
246			conn.commit()
247			cur.close()
248			except (Exception, psycopg2.DatabaseError) as error:
249			return error
250			finally:
251			if conn is not None:
252			conn.close()
253			return user_id
254
255			def request_action(self, status, last_modified, request_id):
256			#PUT /request/<id>/approve
257			sql = """UPDATE tb_request SET status=%s, last_modified=%s WHERE request_id=%s"""
258
259			conn = None
260			try:
261			conn = psycopg2.connect(DATABASE_URL())
262			cur = conn.cursor()
263			cur.execute(sql,(status,last_modified, request_id,))
264
265
266			conn.commit()
267			cur.close()
268
269
270			except (Exception, psycopg2.DatabaseError) as error:
271			return error
272
273			finally:
274			if conn is not None:
275			conn.close()
276

antokish / Maintenance-Tracker

Pull Request — develop (#32)

RequestModel.fetch_for_logged_in_user() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like