build.models.find_by_username() - Code Metrics - Inspection of "Merge pull request #29 from antokish/ft-user-auth-..." - antokish/Maintenance-Tracker - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( 1e8194...5d1e13 )

by Antony

created 2018-06-12 07:37 UTC

build.models.find_by_username() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	24
Code Lines	16

Duplication

Lines	24
Ratio	100 %

Importance

Changes

Metric	Value
eloc	16
dl	24
loc	24
rs	8.9713
c	0
b	0
f	0
cc	3
nop	1

"""
incomplete
implemented using data structures
#creating dtrequest, dtuser and dtlogin list with dictionary 
 to simulate data store

"""

import psycopg2
from config import dbconfig, basedir, filename , section
import os
import jwt

from flask import json, jsonify
from itsdangerous import (TimedJSONWebSignatureSerializer 
                            as Serializer, BadSignature, SignatureExpired)





def test_connection():
    """Test connection to the postgresql server"""

    conn = None
    try:
        #read conection parameters
        params =  dbconfig(filename,section)
        
        #connect to server
        print('Conecting to the PostgreSQL database...')
        conn = psycopg2.connect(**params)
        #create a cursor
        cur = conn.cursor()

        #execute a statement
        print('Postgres database version: ')
        cur.execute('SELECT version()')

        #display the postgress db server version
        db_version = cur.fetchone()
        print(db_version)

        #close comm with pgSQL
        cur.close()
    except (Exception, psycopg2.DatabaseError) as error:
        print(error)
    finally:
        if conn is not None:
            conn.close()
            print('Database connection closed.')






#User sample data
dtusers = [
{
        "id": 1,
        "fname": "John", 
        "lname": "Doe",
        "email": "[email protected]"  
    },
    {
        "id": 2,
        "fname": "Susan", 
        "lname": "Sue",
        "email": "[email protected]" 
    },
    {
        "id": 3,
        "fname": "Mary", 
        "lname": "Doe",
        "email": "[email protected]" 
    },
    {
        "id": 4,
        "fname": "Anto", 
        "lname": "Denis",
        "email": "[email protected]"
    }
]

#requests sample data
dtrequest = [
    {
        "id": 1,
        "requestor":"Anto kish",
        "email": "[email protected]",
        "type": "maintenance",
        "status":"Approved",
        "desc": "Description goes here"
    },
    {
        "id": 2,
        "requestor":"John Doe",
        "email": "[email protected]",
        "type": "repair",
        "status":"Pending",
        "desc": "Description goes here"
    },
    {
        "id": 3,
        "requestor":"Anto kish",
        "email": "[email protected]",
        "type": "maintenance",
        "status":"Pending",
        "desc": "Description goes here"
    },
    {
        "id": 4,
        "requestor":"John Doe",
        "email": "[email protected]",
        "type": "maintenance",
        "status":"Approved",
        "desc": "Description goes here"
    }
]
#login data
dtlogin = [
    {
        "id": 1,
        "username": "[email protected]",
        "password": "pass"
    },
    {
        "id": 2,
        "username": "[email protected]",
        "password": "pass"
    }

]

"""To be implemented when intergrating with database"""
class UserAuth(object):
    def hash_password(self, pwd):
        pass

    def verify_password(self, pwd):
        pass

    def generate_auth_token(self, expiration=600):
        pass

    @staticmethod
    def verify_auth_token(token):
        pass
def find_by_username(username):

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

        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(**params)

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

        result = cur.fetchall()

        cur.close()


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

current_user = find_by_username("[email protected]")

if __name__ == '__main__':
    #test_connection()
    current_user[0]
    print(return_all())

1		"""
2		incomplete
3		implemented using data structures
4		#creating dtrequest, dtuser and dtlogin list with dictionary
5		to simulate data store
6
7		"""
8
9		import psycopg2
10		from config import dbconfig, basedir, filename , section
11		import os
12		import jwt
13
14		from flask import json, jsonify
15		from itsdangerous import (TimedJSONWebSignatureSerializer
16		as Serializer, BadSignature, SignatureExpired)
17
18
19
20
21
22		def test_connection():
23		"""Test connection to the postgresql server"""
24
25		conn = None
26		try:
27		#read conection parameters
28		params = dbconfig(filename,section)
29
30		#connect to server
31		print('Conecting to the PostgreSQL database...')
32		conn = psycopg2.connect(**params)
33		#create a cursor
34		cur = conn.cursor()
35
36		#execute a statement
37		print('Postgres database version: ')
38		cur.execute('SELECT version()')
39
40		#display the postgress db server version
41		db_version = cur.fetchone()
42		print(db_version)
43
44		#close comm with pgSQL
45		cur.close()
46		except (Exception, psycopg2.DatabaseError) as error:
47		print(error)
48		finally:
49		if conn is not None:
50		conn.close()
51		print('Database connection closed.')
52
53
54
55
56
57
58		#User sample data
59		dtusers = [
60		{
61		"id": 1,
62		"fname": "John",
63		"lname": "Doe",
64		"email": "[email protected]"
65		},
66		{
67		"id": 2,
68		"fname": "Susan",
69		"lname": "Sue",
70		"email": "[email protected]"
71		},
72		{
73		"id": 3,
74		"fname": "Mary",
75		"lname": "Doe",
76		"email": "[email protected]"
77		},
78		{
79		"id": 4,
80		"fname": "Anto",
81		"lname": "Denis",
82		"email": "[email protected]"
83		}
84		]
85
86		#requests sample data
87		dtrequest = [
88		{
89		"id": 1,
90		"requestor":"Anto kish",
91		"email": "[email protected]",
92		"type": "maintenance",
93		"status":"Approved",
94		"desc": "Description goes here"
95		},
96		{
97		"id": 2,
98		"requestor":"John Doe",
99		"email": "[email protected]",
100		"type": "repair",
101		"status":"Pending",
102		"desc": "Description goes here"
103		},
104		{
105		"id": 3,
106		"requestor":"Anto kish",
107		"email": "[email protected]",
108		"type": "maintenance",
109		"status":"Pending",
110		"desc": "Description goes here"
111		},
112		{
113		"id": 4,
114		"requestor":"John Doe",
115		"email": "[email protected]",
116		"type": "maintenance",
117		"status":"Approved",
118		"desc": "Description goes here"
119		}
120		]
121		#login data
122		dtlogin = [
123		{
124		"id": 1,
125		"username": "[email protected]",
126		"password": "pass"
127		},
128		{
129		"id": 2,
130		"username": "[email protected]",
131		"password": "pass"
132		}
133
134		]
135
136		"""To be implemented when intergrating with database"""
137		class UserAuth(object):
138		def hash_password(self, pwd):
139		pass
140
141		def verify_password(self, pwd):
142		pass
143
144		def generate_auth_token(self, expiration=600):
145		pass
146
147		@staticmethod
148		def verify_auth_token(token):
149		pass
150	View Code Duplication	def find_by_username(username):
		0 ignored issues – show Duplication introduced 2018-06-11 17:53 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
151		query = """SELECT username,password FROM tb_users WHERE username=(%s)"""
152
153		conn = None
154		result = None
155		try:
156		params = dbconfig(filename, section)
157		conn = psycopg2.connect(**params)
158
159		cur = conn.cursor()
160		cur.execute(query,(username,))
161
162		result = cur.fetchone()
163
164		#print(result)
165
166		cur.close()
167
168		except (Exception, psycopg2.DatabaseError) as error:
169		print(error)
170		finally:
171		if conn is not None:
172		conn.close()
173		return result
174
175	View Code Duplication	def return_all():
		0 ignored issues – show Duplication introduced 2018-06-11 17:53 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
176		query = """select array_to_json(array_agg(row_to_json(t))) from (
177		SELECT * FROM tb_users) t"""
178
179		conn = None
180		result = None
181
182		try:
183		params = dbconfig(filename, section)
184
185		conn = psycopg2.connect(**params)
186
187		cur = conn.cursor()
188		cur.execute(query)
189
190		result = cur.fetchall()
191
192		cur.close()
193
194
195		except (Exception, psycopg2.DatabaseError) as error:
196		print (error)
197		finally:
198		if conn is not None:
199		conn.close()
200		return result
201
202		current_user = find_by_username("[email protected]")
203
204		if __name__ == '__main__':
205		#test_connection()
206		current_user[0]
207		print(return_all())

antokish / Maintenance-Tracker

Push — develop ( 1e8194...5d1e13 )

build.models.find_by_username() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like