Jun-02-2023, 02:22 PM
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)
Keep all python files in same directory:
DbModel.py
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
- DbModel.py
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:
- Create database -- run DbModel
python DbModel.py
(one time)
- Load the database -- run LoadDb
python LoadDb.py
- 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