build.tests.models.create_tables() - 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.tests.models.create_tables() B

↳ Parent: build.tests.models

Complexity

Conditions

Size

Total Lines	66
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	18
dl	0
loc	66
rs	8.8291
c	0
b	0
f	0
cc	4
nop	0

How to fix Long Method

#models tests

import flask
import datetime
import psycopg2

from passlib.hash import pbkdf2_sha256 as sha256
from run import create_app
from flask_bcrypt import Bcrypt




def create_tables():
    """create tables in postgresql database"""
    commands = (
        """
        DROP TABLE IF EXISTS tb_users;
        CREATE TABLE tb_users(
            user_id SERIAL PRIMARY KEY,
            username VARCHAR(100) UNIQUE NOT NULL,
            firstname VARCHAR(50) NOT NULL,
            lastname VARCHAR(50) NOT NULL,
            password VARCHAR(255) NOT NULL,
            created_on TIMESTAMP NOT NULL,
            last_login TIMESTAMP
        )
        """
        ,
        """
        DROP TABLE IF EXISTS tb_request;
        CREATE TABLE tb_request(
            request_id SERIAL PRIMARY KEY,
            requestor VARCHAR(255) NOT NULL,
            email VARCHAR(100) NOT NULL,
            type VARCHAR(50) NOT NULL,
            status VARCHAR(50) NOT NULL,
            description TEXT,
            created_on TIMESTAMP NOT NULL
        )
        """
        ,
        """
        DROP TABLE IF EXISTS tb_admin;
        CREATE TABLE tb_admin(
            admin_id SERIAL PRIMARY KEY,
            username VARCHAR(100) UNIQUE NOT NULL,
            fullname VARCHAR(100) NOT NULL,
            password VARCHAR(255) NOT NULL,
            created_on TIMESTAMP NOT NULL,
            last_login TIMESTAMP
        )
        """
    )

    conn = None
    try:
        
        conn = psycopg2.connect(dbname='test_db',user="antonio", password="pass.123")

        #create cursor
        cur = conn.cursor()
        
        #execute statement
        for command in commands:
            cur.execute(command)

        cur.close()

        #commit the changes
        conn.commit()

    except (Exception, psycopg2.DatabaseError) as error:
        print(error)
    finally:
        if conn is not None:
            conn.close()
    
    return "Tables user, request and admin created succesfully"


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

    query = """INSERT INTO tb_users (username, password, firstname, lastname, created_on)
                VALUES(%s,%s,%s,%s,%s)"""
    
    conn = None
    user_id = None
    try:
        
        conn = psycopg2.connect(dbname='test_db',user="antonio", password="pass.123")

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

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

def find_by_username(username):
    query = """SELECT username,password FROM tb_users WHERE username=(%s)"""
    
    conn = None
    result = None
    try:
        conn = psycopg2.connect(dbname='test_db',user="antonio", password="pass.123")

        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:
        conn = psycopg2.connect(dbname='test_db',user="antonio", password="pass.123")

        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


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)

class RevokedTokenModel(object):
    pass

1			#models tests
2
3			import flask
4			import datetime
5			import psycopg2
6
7			from passlib.hash import pbkdf2_sha256 as sha256
8			from run import create_app
9			from flask_bcrypt import Bcrypt
10
11
12
13
14			def create_tables():
15			"""create tables in postgresql database"""
16			commands = (
17			"""
18			DROP TABLE IF EXISTS tb_users;
19			CREATE TABLE tb_users(
20			user_id SERIAL PRIMARY KEY,
21			username VARCHAR(100) UNIQUE NOT NULL,
22			firstname VARCHAR(50) NOT NULL,
23			lastname VARCHAR(50) NOT NULL,
24			password VARCHAR(255) NOT NULL,
25			created_on TIMESTAMP NOT NULL,
26			last_login TIMESTAMP
27			)
28			"""
29			,
30			"""
31			DROP TABLE IF EXISTS tb_request;
32			CREATE TABLE tb_request(
33			request_id SERIAL PRIMARY KEY,
34			requestor VARCHAR(255) NOT NULL,
35			email VARCHAR(100) NOT NULL,
36			type VARCHAR(50) NOT NULL,
37			status VARCHAR(50) NOT NULL,
38			description TEXT,
39			created_on TIMESTAMP NOT NULL
40			)
41			"""
42			,
43			"""
44			DROP TABLE IF EXISTS tb_admin;
45			CREATE TABLE tb_admin(
46			admin_id SERIAL PRIMARY KEY,
47			username VARCHAR(100) UNIQUE NOT NULL,
48			fullname VARCHAR(100) NOT NULL,
49			password VARCHAR(255) NOT NULL,
50			created_on TIMESTAMP NOT NULL,
51			last_login TIMESTAMP
52			)
53			"""
54			)
55
56			conn = None
57			try:
58
59			conn = psycopg2.connect(dbname='test_db',user="antonio", password="pass.123")
60
61			#create cursor
62			cur = conn.cursor()
63
64			#execute statement
65			for command in commands:
66			cur.execute(command)
67
68			cur.close()
69
70			#commit the changes
71			conn.commit()
72
73			except (Exception, psycopg2.DatabaseError) as error:
74			print(error)
75			finally:
76			if conn is not None:
77			conn.close()
78
79			return "Tables user, request and admin created succesfully"
80
81
82			def insert_to_db(self, username, password, firstname, lastname, created_on):
83			"""insert a new user into database"""
84
85			query = """INSERT INTO tb_users (username, password, firstname, lastname, created_on)
86			VALUES(%s,%s,%s,%s,%s)"""
87
88			conn = None
89			user_id = None
90			try:
91
92			conn = psycopg2.connect(dbname='test_db',user="antonio", password="pass.123")
93
94			cur = conn.cursor()
95			cur.execute(query,(username, password, firstname,
96			lastname, created_on,))
97
98
99			conn.commit()
100			user_id = cur.fetchone()[0]
101			cur.close()
102			except (Exception, psycopg2.DatabaseError) as error:
103			print(error)
104			finally:
105			if conn is not None:
106			conn.close()
107			return user_id
108
109			def find_by_username(username):
110			query = """SELECT username,password FROM tb_users WHERE username=(%s)"""
111
112			conn = None
113			result = None
114			try:
115			conn = psycopg2.connect(dbname='test_db',user="antonio", password="pass.123")
116
117			cur = conn.cursor()
118			cur.execute(query,(username,))
119
120			result = cur.fetchone()
121
122			#print(result)
123
124			cur.close()
125
126			except (Exception, psycopg2.DatabaseError) as error:
127			print(error)
128			finally:
129			if conn is not None:
130			conn.close()
131			return result
132
133			def return_all():
134			query = """select array_to_json(array_agg(row_to_json(t))) from (
135			SELECT * FROM tb_users) t"""
136
137			conn = None
138			result = None
139
140			try:
141			conn = psycopg2.connect(dbname='test_db',user="antonio", password="pass.123")
142
143			cur = conn.cursor()
144			cur.execute(query)
145
146			result = cur.fetchall()
147
148			cur.close()
149
150
151			except (Exception, psycopg2.DatabaseError) as error:
152			print (error)
153			finally:
154			if conn is not None:
155			conn.close()
156			return result
157
158
159			def hash_password(password):
160			#generate hashed string to store in db
161			return sha256.hash(password)
162
163
164			def verify_hash(password, hash):
165			#check the given pass
166			return sha256.verify(password, hash)
167
168			class RevokedTokenModel(object):
169			pass

antokish / Maintenance-Tracker

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

build.tests.models.create_tables() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like