| 171 | |
| 172 | |
| 173 | def _test_dbapi_raw(n, make_objects): |
| 174 | compiled = ( |
| 175 | Customer.__table__.select() |
| 176 | .limit(n) |
| 177 | .compile( |
| 178 | dialect=engine.dialect, compile_kwargs={"literal_binds": True} |
| 179 | ) |
| 180 | ) |
| 181 | |
| 182 | if make_objects: |
| 183 | # because if you're going to roll your own, you're probably |
| 184 | # going to do this, so see how this pushes you right back into |
| 185 | # ORM land anyway :) |
| 186 | class SimpleCustomer: |
| 187 | def __init__(self, id_, name, description): |
| 188 | self.id_ = id_ |
| 189 | self.name = name |
| 190 | self.description = description |
| 191 | |
| 192 | sql = str(compiled) |
| 193 | |
| 194 | conn = engine.raw_connection() |
| 195 | cursor = conn.cursor() |
| 196 | cursor.execute(sql) |
| 197 | |
| 198 | if make_objects: |
| 199 | for row in cursor.fetchall(): |
| 200 | # ensure that we fully fetch! |
| 201 | SimpleCustomer(id_=row[0], name=row[1], description=row[2]) |
| 202 | else: |
| 203 | for row in cursor.fetchall(): |
| 204 | # ensure that we fully fetch! |
| 205 | row[0], row[1], row[2] |
| 206 | |
| 207 | conn.close() |
| 208 | |
| 209 | |
| 210 | if __name__ == "__main__": |