(cls, metadata)
| 51 | |
| 52 | @classmethod |
| 53 | def define_tables(cls, metadata): |
| 54 | global weather_locations |
| 55 | |
| 56 | cls.tables.ids = ids = Table( |
| 57 | "ids", metadata, Column("nextid", Integer, nullable=False) |
| 58 | ) |
| 59 | |
| 60 | def id_generator(ctx): |
| 61 | # in reality, might want to use a separate transaction for this. |
| 62 | |
| 63 | with db1.begin() as c: |
| 64 | nextid = c.execute(ids.select().with_for_update()).scalar() |
| 65 | c.execute( |
| 66 | ids.update().values({ids.c.nextid: ids.c.nextid + 1}) |
| 67 | ) |
| 68 | return nextid |
| 69 | |
| 70 | cls.tables.weather_locations = weather_locations = Table( |
| 71 | "weather_locations", |
| 72 | metadata, |
| 73 | Column("id", Integer, primary_key=True, default=id_generator), |
| 74 | Column("continent", String(30), nullable=False), |
| 75 | Column("city", String(50), nullable=False), |
| 76 | schema=cls.schema, |
| 77 | ) |
| 78 | |
| 79 | cls.tables.weather_reports = Table( |
| 80 | "weather_reports", |
| 81 | metadata, |
| 82 | Column("id", Integer, primary_key=True), |
| 83 | Column("location_id", Integer, ForeignKey(weather_locations.c.id)), |
| 84 | Column("temperature", Numeric(asdecimal=False)), |
| 85 | Column("report_time", DateTime, default=datetime.datetime.now), |
| 86 | schema=cls.schema, |
| 87 | ) |
| 88 | |
| 89 | def setup_test(self): |
| 90 | global db1, db2, db3, db4 |
nothing calls this directly
no test coverage detected