Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[DRAFT] Add testable documentation for regular SQLAlchemy CRUD operations #531

Closed
wants to merge 1 commit into from
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
170 changes: 170 additions & 0 deletions docs/by-example/sqlalchemy/crud.rst
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,58 @@ Create

Insert a new location:

>>> location = Location()
>>> location.name = 'Earth'
>>> location.driver = 'Planet'
>>> location.flag = True

>>> session.add(location)
>>> session.flush()

Refresh "locations" table:

>>> _ = connection.execute(text("REFRESH TABLE locations"))

Inserted location is available:

>>> location = session.query(Location).filter_by(name='Earth').one()
>>> location.name
'Earth'

Retrieve the location from the database:

>>> session.refresh(location)
>>> location.name
'Earth'

Three

>>> location = Location()
>>> location.name = 'Earth'
>>> location.driver = 'Planet'
>>> location.flag = True

>>> session.add(location)
>>> session.flush()

Refresh "locations" table:

>>> _ = connection.execute(text("REFRESH TABLE locations"))

Inserted location is available:

>>> location = session.query(Location).filter_by(name='Earth').one()
>>> location.name
'Earth'

Retrieve the location from the database:

>>> session.refresh(location)
>>> location.name
'Earth'

Three

>>> location = Location()
>>> location.name = 'Earth'
>>> location.kind = 'Planet'
Expand Down Expand Up @@ -204,6 +256,124 @@ Update a record using SQL:
... result.rowcount
1

Update multiple records:

>>> for x in range(10):
... loc = Location()
... loc.name = 'Ort %d' % x
... loc.driver = 'Update'
... session.add(loc)
>>> session.flush()

Refresh table:

>>> _ = connection.execute(text("REFRESH TABLE locations"))

Update multiple records using SQL:

>>> with engine.begin() as conn:
... result = conn.execute(text("update locations set flag=true where kind='Update'"))
... result.rowcount
10

Update all records using SQL, and check that the number of documents affected
of an update without

>>> location = session.query(Location).filter_by(name='Earth').one()

The datetime and date can be set using an update statement:

>>> location.nullable_date = datetime.utcnow().date()
>>> location.nullable_datetime = datetime.utcnow()
>>> session.flush()

Refresh "locations" table:

>>> _ = connection.execute(text("REFRESH TABLE locations"))

Boolean values get set natively:

>>> location.flag
True

Reload the object from the database:

>>> session.refresh(location)

And verify that the date and datetime was persisted:

>>> location.nullable_datetime is not None
True

>>> location.nullable_date is not None
True

Update a record using SQL:

>>> with engine.begin() as conn:
... result = conn.execute(text("update locations set kind='Heimat' where name='Earth'"))
... result.rowcount
1

Update multiple records:

>>> for x in range(10):
... loc = Location()
... loc.name = 'Ort %d' % x
... loc.driver = 'Update'
... session.add(loc)
>>> session.flush()

Refresh table:

>>> _ = connection.execute(text("REFRESH TABLE locations"))

Update multiple records using SQL:

>>> with engine.begin() as conn:
... result = conn.execute(text("update locations set flag=true where kind='Update'"))
... result.rowcount
10

Update all records using SQL, and check that the number of documents affected
of an update without

>>> location = session.query(Location).filter_by(name='Earth').one()

The datetime and date can be set using an update statement:

>>> location.nullable_date = datetime.utcnow().date()
>>> location.nullable_datetime = datetime.utcnow()
>>> session.flush()

Refresh "locations" table:

>>> _ = connection.execute(text("REFRESH TABLE locations"))

Boolean values get set natively:

>>> location.flag
True

Reload the object from the database:

>>> session.refresh(location)

And verify that the date and datetime was persisted:

>>> location.nullable_datetime is not None
True

>>> location.nullable_date is not None
True

Update a record using SQL:

>>> with engine.begin() as conn:
... result = conn.execute(text("update locations set kind='Heimat' where name='Earth'"))
... result.rowcount
1

Update multiple records:

>>> for x in range(10):
Expand Down