SQLAlchemy ORM: mapped classes, Session, add, query, delete.
(r)
| 218 | |
| 219 | |
| 220 | def test_orm(r): |
| 221 | """SQLAlchemy ORM: mapped classes, Session, add, query, delete.""" |
| 222 | print("\n=== ORM ===") |
| 223 | r.set_suite("orm") |
| 224 | engine = create_engine(get_url()) |
| 225 | |
| 226 | try: |
| 227 | with engine.connect() as conn: |
| 228 | conn.execute(text("DROP TABLE IF EXISTS sa_orm_test")) |
| 229 | conn.commit() |
| 230 | |
| 231 | Base.metadata.create_all(engine) |
| 232 | r.ok("ORM create_all") |
| 233 | except Exception as e: |
| 234 | r.fail("ORM create_all", str(e)) |
| 235 | engine.dispose() |
| 236 | return |
| 237 | |
| 238 | try: |
| 239 | with Session(engine) as session: |
| 240 | session.add_all([ |
| 241 | OrmTest(id=1, name="alice", value=3.14), |
| 242 | OrmTest(id=2, name="bob", value=2.72), |
| 243 | ]) |
| 244 | session.commit() |
| 245 | r.ok("ORM add_all", "2 rows") |
| 246 | |
| 247 | results = session.query(OrmTest).order_by(OrmTest.id).all() |
| 248 | if len(results) == 2 and results[0].name == "alice": |
| 249 | r.ok("ORM query", f"{len(results)} rows") |
| 250 | else: |
| 251 | r.fail("ORM query", f"unexpected: {results}") |
| 252 | |
| 253 | obj = session.get(OrmTest, 1) |
| 254 | if obj and obj.name == "alice": |
| 255 | r.ok("ORM get") |
| 256 | else: |
| 257 | r.fail("ORM get", f"unexpected: {obj}") |
| 258 | |
| 259 | obj.value = 9.99 |
| 260 | session.commit() |
| 261 | refreshed = session.get(OrmTest, 1) |
| 262 | if refreshed and refreshed.value == 9.99: |
| 263 | r.ok("ORM update") |
| 264 | else: |
| 265 | r.fail("ORM update", f"value: {refreshed.value if refreshed else 'None'}") |
| 266 | |
| 267 | bob = session.get(OrmTest, 2) |
| 268 | if bob: |
| 269 | session.delete(bob) |
| 270 | session.commit() |
| 271 | r.ok("ORM delete") |
| 272 | else: |
| 273 | r.fail("ORM delete", "could not find id=2") |
| 274 | |
| 275 | remaining = session.query(OrmTest).count() |
| 276 | if remaining == 1: |
| 277 | r.ok("ORM post-delete count", "1 row") |