Merge pull request #34 from yankeexe/hint-dt-mssql

Basic type hinting for functions and global variables

examples/app-mssql/main.py

@@ -1,25 +1,26 @@
 ''' Main module. '''
-import sys
 import os
+import sys
 import time
+from typing import List, Tuple
+
 import pyodbc
 from faker import Faker
-from typing import List, Tuple
 
 
-CONNECTION_STRING = 'DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={server};DATABASE={database};UID={username};PWD={password};'
+CONNECTION_STRING: str = 'DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={server};DATABASE={database};UID={username};PWD={password};'
 
-SQL_CREATE_TABLE = '''
+SQL_CREATE_TABLE: str = '''
     CREATE TABLE users (
         id INT,
         name VARCHAR(50),
         city VARCHAR(50)
     )
 '''
 
-SQL_INSERT_DATA = 'INSERT INTO users (id, name, city) VALUES (?, ?, ?)'
+SQL_INSERT_DATA: str = 'INSERT INTO users (id, name, city) VALUES (?, ?, ?)'
 
-RECORD_COUNT = 10000
+RECORD_COUNT: int = 10000
 
 
 def get_data(count: int) -> List[Tuple]:
@@ -30,7 +31,7 @@ def get_data(count: int) -> List[Tuple]:
     return [row(i) for i in range(count)]
 
 
-def connect_db():
+def connect_db() -> pyodbc.Connection:
     ''' Connect to database. '''
     print('Establishing mssql database connection.')
     connection_str = CONNECTION_STRING.format(
@@ -43,7 +44,7 @@ def connect_db():
     return pyodbc.connect(connection_str, timeout=300)
 
 
-def setup_table(cur, data):
+def setup_table(cur: pyodbc.Cursor, data: List):
     ''' Create table and populate data. '''
     print('Create a new table for users.')
     cur.execute(SQL_CREATE_TABLE)
@@ -55,15 +56,15 @@ def setup_table(cur, data):
     cur.commit()
 
 
-def fetch_data(cur):
+def fetch_data(cur: pyodbc.Cursor) -> List:
     ''' Fetch all data from the table. '''
     print('List of data.')
     cur.execute('SELECT * FROM users')
 
     return cur.fetchall()
 
 
-def display_data(rows):
+def display_data(rows: List[Tuple[int, str, str]]):
     ''' Print rows in the console. '''
     template = '{:<5} {:<15} {:<10}'
     print(template.format('ID', 'NAME', 'CITY'))
@@ -79,7 +80,6 @@ def main():
     cur = conn.cursor()
     data = get_data(RECORD_COUNT)
 
-
     setup_table(cur, data)
     rows = fetch_data(cur)
     display_data(rows)

examples/app-pg/main.py

@@ -1,25 +1,26 @@
 ''' Main module. '''
-import sys
 import os
+import sys
 import time
+from typing import List, Tuple
+
 import pyodbc
 from faker import Faker
-from typing import List, Tuple
 
 
-CONNECTION_STRING = 'DRIVER={{PostgreSQL Unicode}};SERVER={server};DATABASE={database};UID={username};PWD={password};'
+CONNECTION_STRING: str = 'DRIVER={{PostgreSQL Unicode}};SERVER={server};DATABASE={database};UID={username};PWD={password};'
 
-SQL_CREATE_TABLE = '''
+SQL_CREATE_TABLE: str = '''
     CREATE TABLE users (
         id INT,
         name VARCHAR(50),
         city VARCHAR(50)
     )
 '''
 
-SQL_INSERT_DATA = 'INSERT INTO users (id, name, city) VALUES (?, ?, ?)'
+SQL_INSERT_DATA: str = 'INSERT INTO users (id, name, city) VALUES (?, ?, ?)'
 
-RECORD_COUNT = 10000
+RECORD_COUNT: int = 10000
 
 
 def get_data(count: int) -> List[Tuple]:
@@ -30,7 +31,7 @@ def get_data(count: int) -> List[Tuple]:
     return [row(i) for i in range(count)]
 
 
-def connect_db():
+def connect_db() -> pyodbc.Connection:
     ''' Connect to database. '''
     print('Establishing pg database connection.')
     connection_str = CONNECTION_STRING.format(
@@ -43,7 +44,7 @@ def connect_db():
     return pyodbc.connect(connection_str, timeout=300)
 
 
-def setup_table(cur, data):
+def setup_table(cur: pyodbc.Cursor, data: List):
     ''' Create table and populate data. '''
     print('Create a new table for users.')
     cur.execute(SQL_CREATE_TABLE)
@@ -55,15 +56,15 @@ def setup_table(cur, data):
     cur.commit()
 
 
-def fetch_data(cur):
+def fetch_data(cur: pyodbc.Cursor) -> List:
     ''' Fetch all data from the table. '''
     print('List of data.')
     cur.execute('SELECT * FROM users')
 
     return cur.fetchall()
 
 
-def display_data(rows):
+def display_data(rows: List[Tuple[int, str, str]]):
     ''' Print rows in the console. '''
     template = '{:<5} {:<15} {:<10}'
     print(template.format('ID', 'NAME', 'CITY'))

examples/data-transfer-mssql/main.py

@@ -1,13 +1,19 @@
-import sys
+''' Main module. '''
 import os
+import sys
 import time
+from typing import Tuple
+
 import pyodbc
 from faker import Faker
 
 
-CONNECTION_STRING = 'DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={server};DATABASE={database};UID={username};PWD={password};'
-RECORD_COUNT = 10000
-SQL_INSERT_DATA = 'INSERT INTO users (id, name, city) VALUES (?, ?, ?);'
+CONNECTION_STRING: str = 'DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={server};DATABASE={database};UID={username};PWD={password};'
+
+RECORD_COUNT: int = 10000
+
+SQL_INSERT_DATA: str = 'INSERT INTO users (id, name, city) VALUES (?, ?, ?);'
+
 
 
 def main():
@@ -37,7 +43,7 @@ def main():
             display_users(dest_db_cur)
 
 
-def connect_to_databases():
+def connect_to_databases() -> Tuple:
     ''' Extracts databases credentials from the environment and returns their connections. '''
     source_db_conn = get_connection(
         os.environ['SOURCE_DB_HOST'],
@@ -56,7 +62,7 @@ def connect_to_databases():
     return source_db_conn, dest_db_conn
 
 
-def get_connection(db_host, db_name, db_username, db_password):
+def get_connection(db_host: str, db_name: str, db_username: str, db_password: str) -> pyodbc.Connection:
     ''' Create database connection and returns connection. '''
     connection_str = CONNECTION_STRING.format(
         server=db_host,
@@ -68,7 +74,7 @@ def get_connection(db_host, db_name, db_username, db_password):
     return pyodbc.connect(connection_str, timeout=300)
 
 
-def populate_data(RECORD_COUNT: int, db_cursor):
+def populate_data(RECORD_COUNT: int, db_cursor: pyodbc.Cursor):
     ''' Generate user data. '''
     fake = Faker()
     row = lambda n: (n + 1, fake.format('name'), fake.format('city'))
@@ -77,15 +83,15 @@ def populate_data(RECORD_COUNT: int, db_cursor):
         db_cursor.execute(SQL_INSERT_DATA, row(i))
 
 
-def extract_sql(file: str):
+def extract_sql(file: str) -> str:
     ''' Reads an SQL file and returns it's contents. '''
     with open(file, 'rt') as file:
         contents = file.read()
 
     return contents
 
 
-def transfer_data(source_db_cursor, dest_db_cursor, dest_db_conn):
+def transfer_data(source_db_cursor: pyodbc.Cursor, dest_db_cursor: pyodbc.Cursor, dest_db_conn: pyodbc.Connection):
     ''' Extracts users data from source database and stores them in destination database. '''
     print('Extracting users data from source database.')
     source_db_cursor.execute('SELECT * FROM users')
@@ -99,7 +105,7 @@ def transfer_data(source_db_cursor, dest_db_cursor, dest_db_conn):
     dest_db_conn.commit()
 
 
-def display_users(db_cursor):
+def display_users(db_cursor: pyodbc.Cursor):
     ''' Displays users data. '''
     db_cursor.execute('SELECT * FROM users ORDER BY id')
     transferred_data = db_cursor.fetchall()

examples/data-transfer-pg/main.py

@@ -1,15 +1,18 @@
 ''' Main module. '''
-import sys
 import os
+import sys
 import time
+from typing import Tuple
+
 import pyodbc
 from faker import Faker
-from typing import List, Tuple
 
 
-CONNECTION_STRING = 'DRIVER={{PostgreSQL Unicode}};SERVER={server};DATABASE={database};UID={username};PWD={password};'
-RECORD_COUNT = 10000
-SQL_INSERT_DATA = 'INSERT INTO users (id, name, city) VALUES (?, ?, ?);'
+CONNECTION_STRING: str = 'DRIVER={{PostgreSQL Unicode}};SERVER={server};DATABASE={database};UID={username};PWD={password};'
+
+RECORD_COUNT: int = 10000
+
+SQL_INSERT_DATA: str = 'INSERT INTO users (id, name, city) VALUES (?, ?, ?);'
 
 
 def main():
@@ -42,7 +45,7 @@ def main():
             display_users(dest_db_cur)
 
 
-def get_connection(host: str, db_name: str, db_user: str, db_password: str):
+def get_connection(host: str, db_name: str, db_user: str, db_password: str) -> pyodbc.Connection:
     ''' Initiates and returns connection of a database.'''
     print(f'Establishing postgres database connection to {host}.')
     connection_str = CONNECTION_STRING.format(
@@ -51,10 +54,11 @@ def get_connection(host: str, db_name: str, db_user: str, db_password: str):
         username=db_user,
         password=db_password
     )
+
     return pyodbc.connect(connection_str, timeout=300)
 
 
-def connect_to_databases():
+def connect_to_databases() -> Tuple:
     ''' Extracts databases credentials from the environment and returns their connections.'''
     source_db_conn = get_connection(
         os.environ['DB1_HOST'],
@@ -73,7 +77,7 @@ def connect_to_databases():
     return source_db_conn, dest_db_conn
 
 
-def populate_data(count: int, db_cursor):
+def populate_data(count: int, db_cursor: pyodbc.Cursor):
     ''' Generate user data. '''
     fake = Faker()
     row = lambda n: (n + 1, fake.format('name'), fake.format('city'))
@@ -82,15 +86,15 @@ def populate_data(count: int, db_cursor):
         db_cursor.execute(SQL_INSERT_DATA, row(i))
 
 
-def extract_sql(file: str):
+def extract_sql(file: str) -> str:
     ''' Reads an SQL file and returns it's contents.'''
     with open(file, 'rt') as file:
         contents = file.read()
 
     return contents
 
 
-def transfer_data(source_db_cursor, dest_db_cursor, dest_db_conn):
+def transfer_data(source_db_cursor: pyodbc.Cursor, dest_db_cursor: pyodbc.Cursor, dest_db_conn: pyodbc.Connection):
     '''
     Extracts users data from source database and 
     stores them in destination database.
@@ -104,12 +108,10 @@ def transfer_data(source_db_cursor, dest_db_cursor, dest_db_conn):
         dest_db_cursor.execute(SQL_INSERT_DATA, (row.id, row.name, row.city))
     dest_db_conn.commit()
 
-    print(
-        f'Transferred {len(rows)} rows of users data from source database to destination database.'
-    )
+    print(f'Transferred {len(rows)} rows of users data from source database to destination database.')
 
 
-def display_users(db_cursor):
+def display_users(db_cursor: pyodbc.Cursor):
     ''' Displays users data. '''
     db_cursor.execute('SELECT * FROM users')
     transferred_data = db_cursor.fetchall()

test/test_mssql.py

@@ -1,5 +1,4 @@
 ''' Tests for MSSQL. '''
-
 from util import exec_query, MSSQL
 
 

test/test_mysql.py

@@ -1,5 +1,4 @@
 ''' Tests for MySQL. '''
-
 from util import exec_query, MYSQL
 
 

test/test_pg.py

@@ -1,5 +1,4 @@
 ''' Tests for PostgreSQL. '''
-
 from util import exec_query, PG