Database Programming is Program with Data

Schema of Users table in Sqlite.db

Uses PRAGMA statement to read schema.

Describe Schema, here is resource Resource- What is a database schema?

  • What is the purpose of identity Column in SQL database?
  • What is the purpose of a primary key in SQL database?
  • What are the Data Types in SQL table?
import sqlite3

database = 'instance/sqlite.db' # this is location of database

def schema():
    
    # Connect to the database file
    conn = sqlite3.connect(database)

    # Create a cursor object to execute SQL queries
    cursor = conn.cursor()
    
    # Fetch results of Schema
    results = cursor.execute("PRAGMA table_info('users')").fetchall()

    # Print the results
    for row in results:
        print(row)

    # Close the database connection
    conn.close()
    
schema()
Traceback (most recent call last):
  File "_pydevd_bundle/pydevd_cython.pyx", line 1078, in _pydevd_bundle.pydevd_cython.PyDBFrame.trace_dispatch
  File "_pydevd_bundle/pydevd_cython.pyx", line 297, in _pydevd_bundle.pydevd_cython.PyDBFrame.do_wait_suspend
  File "/Users/lukeriggins/opt/anaconda3/lib/python3.9/site-packages/debugpy/_vendored/pydevd/pydevd.py", line 1976, in do_wait_suspend
    keep_suspended = self._do_wait_suspend(thread, frame, event, arg, suspend_type, from_this_thread, frames_tracker)
  File "/Users/lukeriggins/opt/anaconda3/lib/python3.9/site-packages/debugpy/_vendored/pydevd/pydevd.py", line 2011, in _do_wait_suspend
    time.sleep(0.01)
KeyboardInterrupt
---------------------------------------------------------------------------
KeyboardInterrupt                         Traceback (most recent call last)
/var/folders/rb/cg1zlcqx3kd5jwkb_g7zw1980000gn/T/ipykernel_73546/27105271.py in <module>
     21     conn.close()
     22 
---> 23 schema()

/var/folders/rb/cg1zlcqx3kd5jwkb_g7zw1980000gn/T/ipykernel_73546/27105271.py in schema()
      6 
      7     # Connect to the database file
----> 8     conn = sqlite3.connect(database)
      9 
     10     # Create a cursor object to execute SQL queries

/var/folders/rb/cg1zlcqx3kd5jwkb_g7zw1980000gn/T/ipykernel_73546/27105271.py in schema()
      6 
      7     # Connect to the database file
----> 8     conn = sqlite3.connect(database)
      9 
     10     # Create a cursor object to execute SQL queries

_pydevd_bundle/pydevd_cython.pyx in _pydevd_bundle.pydevd_cython.SafeCallWrapper.__call__()

_pydevd_bundle/pydevd_cython.pyx in _pydevd_bundle.pydevd_cython.PyDBFrame.trace_dispatch()

_pydevd_bundle/pydevd_cython.pyx in _pydevd_bundle.pydevd_cython.PyDBFrame.trace_dispatch()

_pydevd_bundle/pydevd_cython.pyx in _pydevd_bundle.pydevd_cython.PyDBFrame.trace_dispatch()

_pydevd_bundle/pydevd_cython.pyx in _pydevd_bundle.pydevd_cython.PyDBFrame.do_wait_suspend()

~/opt/anaconda3/lib/python3.9/site-packages/debugpy/_vendored/pydevd/pydevd.py in do_wait_suspend(self, thread, frame, event, arg, exception_type)
   1974 
   1975             with self._threads_suspended_single_notification.notify_thread_suspended(thread_id, stop_reason):
-> 1976                 keep_suspended = self._do_wait_suspend(thread, frame, event, arg, suspend_type, from_this_thread, frames_tracker)
   1977 
   1978         frames_list = None

~/opt/anaconda3/lib/python3.9/site-packages/debugpy/_vendored/pydevd/pydevd.py in _do_wait_suspend(self, thread, frame, event, arg, suspend_type, from_this_thread, frames_tracker)
   2009 
   2010             self.process_internal_commands()
-> 2011             time.sleep(0.01)
   2012 
   2013         self.cancel_async_evaluation(get_current_thread_id(thread), str(id(frame)))

KeyboardInterrupt: 

Reading Users table

import sqlite3

def read():
    # Connect to the database file
    conn = sqlite3.connect(database)

    # Create a cursor object to execute SQL queries
    cursor = conn.cursor()
    
    # Execute a SELECT statement to retrieve data from a table
    results = cursor.execute('SELECT * FROM users').fetchall()

    # Print the results
    if len(results) == 0:
        print("Table is empty")
    else:
        for row in results:
            print(row)

    # Close the cursor and connection objects
    cursor.close()
    conn.close()
    
read()
---------------------------------------------------------------------------
OperationalError                          Traceback (most recent call last)
/var/folders/rb/cg1zlcqx3kd5jwkb_g7zw1980000gn/T/ipykernel_73546/3854437116.py in <module>
     22     conn.close()
     23 
---> 24 read()

/var/folders/rb/cg1zlcqx3kd5jwkb_g7zw1980000gn/T/ipykernel_73546/3854437116.py in read()
      9 
     10     # Execute a SELECT statement to retrieve data from a table
---> 11     results = cursor.execute('SELECT * FROM users').fetchall()
     12 
     13     # Print the results

OperationalError: no such table: users

Create a new User

import sqlite3

def create():
    name = input("Enter your name:")
    uid = input("Enter your user id:")
    password = input("Enter your password")
    dob = input("Enter your date of birth 'YYYY-MM-DD'")
    
    # Connect to the database file
    conn = sqlite3.connect(database)

    # Create a cursor object to execute SQL commands
    cursor = conn.cursor()

    try:
        # Execute an SQL command to insert data into a table
        cursor.execute("INSERT INTO users (_name, _uid, _password, _dob) VALUES (?, ?, ?, ?)", (name, uid, password, dob))
        
        # Commit the changes to the database
        conn.commit()
        print(f"A new user record {uid} has been created")
                
    except sqlite3.Error as error:
        print("Error while executing the INSERT:", error)


    # Close the cursor and connection objects
    cursor.close()
    conn.close()
    
#create()

Updating a User

import sqlite3

def update():
    uid = input("Enter user id to update")
    password = input("Enter updated password")
    if len(password) < 2:
        message = "hacked"
        password = 'gothackednewpassword123'
    else:
        message = "successfully updated"

    # Connect to the database file
    conn = sqlite3.connect(database)

    # Create a cursor object to execute SQL commands
    cursor = conn.cursor()

    try:
        # Execute an SQL command to update data in a table
        cursor.execute("UPDATE users SET _password = ? WHERE _uid = ?", (password, uid))
        if cursor.rowcount == 0:
            # The uid was not found in the table
            print(f"No uid {uid} was not found in the table")
        else:
            print(f"The row with user id {uid} the password has been {message}")
            conn.commit()
    except sqlite3.Error as error:
        print("Error while executing the UPDATE:", error)
        
    
    # Close the cursor and connection objects
    cursor.close()
    conn.close()
    
#update()

Delete a User in table in Sqlite.db

Uses a delete function to remove a user based on a user input of the id.

  • Is DELETE a dangerous operation? Why?
  • In the print statemements, what is the "f" and what does {uid} do?
import sqlite3

def delete():
    uid = input("Enter user id to delete")

    # Connect to the database file
    conn = sqlite3.connect(database)

    # Create a cursor object to execute SQL commands
    cursor = conn.cursor()
    
    try:
        cursor.execute("DELETE FROM users WHERE _uid = ?", (uid,))
        if cursor.rowcount == 0:
            # The uid was not found in the table
            print(f"No uid {uid} was not found in the table")
        else:
            # The uid was found in the table and the row was deleted
            print(f"The row with uid {uid} was successfully deleted")
        conn.commit()
    except sqlite3.Error as error:
        print("Error while executing the DELETE:", error)
        
    # Close the cursor and connection objects
    cursor.close()
    conn.close()
    
#delete()

Menu Interface to CRUD operations

CRUD and Schema interactions from one location by running menu. Observe input at the top of VSCode, observe output underneath code cell.

  • Why does the menu repeat?
  • Could you refactor this menu? Make it work with a List?
def menu():
    operation = input("Enter: (C)reate (R)ead (U)pdate or (D)elete or (S)chema")
    if operation.lower() == 'c':
        create()
    elif operation.lower() == 'r':
        read()
    elif operation.lower() == 'u':
        update()
    elif operation.lower() == 'd':
        delete()
    elif operation.lower() == 's':
        schema()
    elif len(operation)==0: # Escape Key
        return
    else:
        print("Please enter c, r, u, or d") 
    menu() # recursion, repeat menu
        
try:
    menu() # start menu
except:
    print("Perform Jupyter 'Run All' prior to starting menu")