Wikiversity

Applied Programming/Databases/Python3 SQLAlchemy

< Applied Programming | Databases

databases.py

edit 
"""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 It

edit

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 Also

edit
Retrieved from "https://en.wikiversity.org/w/index.php?title=Applied_Programming/Databases/Python3_SQLAlchemy&oldid=2140384"