Applied Programming/Databases/Python3 SQLAlchemy

databases.pyEdit

"""This program demonstrates SQLAlchemy database processing.

Input:
    None

Output:
    Sample data.

References:
    https://en.wikiversity.org/wiki/Python_Programming/Databases
    
"""

import sqlalchemy
import sys


engine = sqlalchemy.create_engine('sqlite:///users.db')
metadata = sqlalchemy.MetaData()
users = sqlalchemy.Table("Users", metadata,
        sqlalchemy.Column("UserID", sqlalchemy.Integer, primary_key=True),
        sqlalchemy.Column("User", sqlalchemy.Text, nullable=False)
    )


def create_table():
    """Creates the Users table.
    
    Args:
        None.
        
    Returns:
        None.
    
    """
    metadata.create_all(engine)


def insert_users():
    """Inserts data into the Users table.
    
    Args:
        None.
        
    Returns:
        None.
    
    """
    try:
        with engine.connect() as connection:
            insert = users.insert(None).values({"UserID": 1, "User": "Moe"})
            connection.execute(insert)

            insert = users.insert(None).values({"UserID": 2, "User": "Larry"})
            connection.execute(insert)

            insert = users.insert(None).values({"UserID": 3, "User": "Curly"})
            connection.execute(insert)
    except Exception as exception:
        print(exception)


def display_table():
    """Displays the Users table.
    
    Args:
        None.
        
    Returns:
        None.
    
    """
    try:
        with engine.connect() as connection:
            select = users.select()
            cursor = connection.execute(select)
            rows = cursor.fetchall()

            for row in rows:
                print(row)
            print()
    except Exception as exception:
        print(exception)


def update_user():
    """Updates the Users table.
    
    Args:
        None.
        
    Returns:
        None.
    
    """
    try:
        with engine.connect() as connection:
            update = users.update(None).where("UserID=3").values({"User":"Shemp"})
            connection.execute(update)
    except Exception as exception:
        print(exception)


def delete_user():
    """Deletes a record from the Users table.
    
    Args:
        None.
        
    Returns:
        None.
    
    """
    try:
        with engine.connect() as connection:
            delete = users.delete(None).where("UserID=3")
            connection.execute(delete)
    except Exception as exception:
        print(exception)


def main():
    """Runs the main program logic."""

    try:
        print("Users")
        create_table()
        insert_users()
        display_table()
    
        print("Users After Update")
        update_user()
        display_table()
    
        print("Users After Delete")
        delete_user()
        display_table()
    except:
        print("Unexpected error.")
        print("Error:", sys.exc_info()[1])
        print("File: ", sys.exc_info()[2].tb_frame.f_code.co_filename) 
        print("Line: ", sys.exc_info()[2].tb_lineno)
    

main()

Try ItEdit

Copy and paste the code above into one of the following free online development environments or use your own Python3 SQLAlchemy compiler / interpreter / IDE.

See AlsoEdit