Rows not adding to sqlite3 database using SQLAlchemy

[Larz60+]

Here is what i did (and this all works)
NOTE: id is an integer field used (and populated) by sqlalchemy
so I created a new field row_id (see code)

• I split your classes into separate files
  • DbModel.py
    
  • LoadDb.py
    
  • added a query test routine: query.db
    

Keep all python files in same directory:

DbModel.py

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, backref
from sqlalchemy import Column, Integer, String, Date, create_engine
from datetime import datetime


engine = create_engine(f"sqlite:///./test_catalog.db")
Base = declarative_base()


class TestDb(Base):
    
    __tablename__ = "catalog"
 
    # Note: id is an integer field used internally by SQLalchemy
    #       it is not loaded by user.
    id = Column(Integer, primary_key=True)  # Identifying ID for this test
    
    row_id = Column(String) # id for this entry
    parent_id = Column(String)  # ID of test that this was generated from, if not a parent test
    request_id = Column(String)  # UUID ID used to group multiple tests that are submitted at the same time
    test_type = Column(String)  # Type of test: Online, Impaired, Network, etc...
    target = Column(String)  # Target for this test
    user = Column(String)  # Username of who initiated the test
    submitted = Column(Date)  # When was this test submitted
    started = Column(Date)  # When was this test started
    ended = Column(Date)  # When was this test completed
    progress = Column(Integer, default=0)  # Progress indicator for this test, from 0 to 100
    state = Column(String, default="New")  # State of the test: New, Running, Done, Error
    result = Column(String)  # JSON results of the test. Compressed?
 

    def __init__(self, row_id, parent_id, request_id, test_type,
        target, user, submitted, started, ended,
        progress, state, result):
        self.row_id = row_id
        self.parent_id = parent_id
        self.request_id = request_id
        self.test_type = test_type
        self.target = target
        self.user = user
        self.submitted = submitted
        self.started = started
        self.ended = ended
        self.progress = progress
        self.state = state
        self.result = result


def main():
    Base.metadata.create_all(engine)
 
 
if __name__ == "__main__":
    main()

LoadDb.py

from sqlalchemy.orm import sessionmaker
from datetime import datetime
import DbModel

class LoadTables:
    def __init__(self):
        self.dmap = DbModel
        self.db = self.dmap.engine

    def load_catalog(self):
        db = self.db
        Session = sessionmaker(bind=db)
        Session.configure(bind=db)
        session = Session()

        testrec = self.dmap.TestDb(
            row_id = "1234567890",
            parent_id = None,
            request_id = "0987654321",
            target = "target",
            user = "user",
            test_type = "test",
            submitted = datetime.now(),
            started = None,
            ended = None,
            progress = 0,
            state = "New",
            result = None)
        session.add(testrec)

        session.commit()
        session.close()


def main():
    lt = LoadTables()
    lt.load_catalog()
  
  
if __name__ == '__main__':
    main()

QueryDb.py

import DbModel
from sqlalchemy.orm import sessionmaker
 
 
class QueryCityTable:
    def __init__(self):
        self.dmap = DbModel
        db = self.dmap.engine
        self.Session = sessionmaker(bind=db)
        self.Session.configure(bind=db)
        self.session = self.Session()
        self.catalog = self.dmap.TestDb
 
    def show_catalog(self):
        # The query (note you can add filter if needed)
        catalog = self.session.query(self.catalog).all()
        print(f"List of Entire catalog")
        for row in catalog:
            print(f"row_id: {row.row_id}")
            print(f"parent_id: {row.parent_id}")
            print(f"request_id: {row.request_id}")
            print(f"test_type: {row.test_type}")
            print(f"target: {row.target}")
            print(f"user: {row.user}")
            print(f"submitted: {row.submitted}")
            print(f"started: {row.started}")
            print(f"ended: {row.ended}")
            print(f"progress: {row.progress}")
            print(f"state: {row.state}")
            print(f"result: {row.result}")


def main():
    qct = QueryCityTable()
    qct.show_catalog()


if __name__ == '__main__':
    main()

Instructions:

1. Create database -- run DbModel python DbModel.py (one time)
   
2. Load the database -- run LoadDb python LoadDb.py
   
3. Query resyults -- run Query.db python Query.py
   
   results:
   

   Output:
   List of Entire catalog
   row_id: 1234567890
   parent_id: None
   request_id: 0987654321
   test_type: test
   target: target
   user: user
   submitted: 2023-06-02
   started: None
   ended: None
   progress: 0
   state: New
   result: None

[bowlofred]

When I get a successful connection, it will return from the function, so never do the finally block. The db shouldn't close.

No, that's incorrect. The finally block always runs, even if you return out of the try. That's the point of the finally block.

def myfunc():
    try:
        a = 25
        return a
    except Exception as e:
        print (f"Got a problem.  {e}")
    finally:
        print ("Running finally.  Closing DB")

answer = myfunc()
print(f"I got {answer} from the function")

Output:
Running finally.  Closing DB
I got 25 from the function

Why not just put the close in the exception block? What do you expect to happen in finally that wouldn't happen in an exception?