6efisqAT

1#!/usr/bin/python2.7
2#
3# Interface for the assignement
4#
5
6import psycopg2
7from threading import Thread
8from Queue import Queue
9
10DATABASE_NAME = 'dds_assgn1'
11
12q = Queue()
13
14
15def getopenconnection(user='postgres', password='sGurura3', dbname='dds_assgn1'):
16    return psycopg2.connect("dbname='" + dbname + "' user='" + user + "' host='localhost' password='" + password + "'")
17
18
19def load_data_set(line, conn, ratingstablename):
20    cur = conn.cursor()
21    try:
22        user_id, movie_id, rating, timestamp = line.split("::")
23        cur.execute('''INSERT INTO %s VALUES(%d, %d, %f)''' %
24                    (ratingstablename, int(user_id), int(movie_id), float(rating)))
25    except Exception as ex:
26        print 'DB Insert error: ', ex.message
27    conn.commit()
28
29def define_thread():
30    while True:
31        dataset, conn, ratingstablename = q.get()
32        load_data_set(dataset, conn, ratingstablename)
33        q.task_done()
34
35def loadratings(ratingstablename, ratingsfilepath, openconnection):
36    try:
37        for i in range(50):  # 50 worker threads
38            t = Thread(target=define_thread)
39            t.daemon = True
40            t.start()
41        with open(ratingsfilepath) as the_file:
42            line = the_file.readline()
43            while line:
44                q.put((line, openconnection, ratingstablename))
45                line = the_file.readline()
46        q.join()
47    except Exception as ex:
48        print 'Generic exception: ', ex.message
49        raise ex
50
51
52def rangepartition(ratingstablename, numberofpartitions, openconnection):
53    pass
54
55
56def roundrobinpartition(ratingstablename, numberofpartitions, openconnection):
57    pass
58
59
60def roundrobininsert(ratingstablename, userid, itemid, rating, openconnection):
61    pass
62
63
64def rangeinsert(ratingstablename, userid, itemid, rating, openconnection):
65    pass
66
67
68def create_db(dbname):
69    """
70    We create a DB by connecting to the default user and database of Postgres
71    The function first checks if an existing database exists for a given name, else creates it.
72    :return:None
73    """
74    # Connect to the default database
75    con = getopenconnection(dbname='postgres')
76    con.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
77    cur = con.cursor()
78
79    # Check if an existing database with the same name exists
80    cur.execute('SELECT COUNT(*) FROM pg_catalog.pg_database WHERE datname=\'%s\'' % (dbname,))
81    count = cur.fetchone()[0]
82    if count == 0:
83        cur.execute('CREATE DATABASE %s' % (dbname,))  # Create the database
84    else:
85        print 'A database named {0} already exists'.format(dbname)
86
87    # Clean up
88    cur.close()
89    con.close()
90
91
92# Middleware
93def before_db_creation_middleware():
94    # Use it if you want to
95    pass
96
97
98def after_db_creation_middleware(databasename):
99    # Use it if you want to
100    with getopenconnection(dbname=databasename) as conn:
101        cursor = conn.cursor()
102        cursor.execute("CREATE TABLE IF NOT EXISTS Ratings (user_id INTEGER, "
103                       "movie_id INTEGER, "
104                       "rating REAL)")
105
106def before_test_script_starts_middleware(openconnection, databasename):
107    # Use it if you want to
108    pass
109
110
111def after_test_script_ends_middleware(openconnection, databasename):
112    # Use it if you want to
113    pass
114
115
116if __name__ == '__main__':
117    try:
118
119        # Use this function to do any set up before creating the DB, if any
120        before_db_creation_middleware()
121
122        create_db(DATABASE_NAME)
123
124        # Use this function to do any set up after creating the DB, if any
125        after_db_creation_middleware(DATABASE_NAME)
126
127        with getopenconnection() as con:
128            # Use this function to do any set up before I starting calling your functions to test, if you want to
129            before_test_script_starts_middleware(con, DATABASE_NAME)
130
131            # Here is where I will start calling your functions to test them. For example,
132            loadratings('Ratings', 'ratings.dat', con)
133            # ###################################################################################
134            # Anything in this area will not be executed as I will call your functions directly
135            # so please add whatever code you want to add in main, in the middleware functions provided "only"
136            # ###################################################################################
137
138            # Use this function to do any set up after I finish testing, if you want to
139            after_test_script_ends_middleware(con, DATABASE_NAME)
140
141    except Exception as detail:
142        print "OOPS! This is the error ==> ", detail