|
Size: 1370
Comment:
|
Size: 1590
Comment:
|
| Deletions are marked like this. | Additions are marked like this. |
| Line 1: | Line 1: |
| = Python Context Manager = | ## page was renamed from Python/ContextManagers = Python Context Managers = |
| Line 9: | Line 10: |
| == Basic Example == | == Example == |
| Line 11: | Line 12: |
| {{{#!highlight python | {{{ |
| Line 13: | Line 14: |
| import pprint from typing import * |
from pprint import pprint from types import TracebackType |
| Line 17: | Line 18: |
| """A reference to a database.""" | def __init__(self, cursor: sqlite3.Cursor) -> None: self._cursor = cursor |
| Line 19: | Line 21: |
| def __init__( self, cursor: sqlite3.Cursor, ) -> None: self.curs = cursor |
def execute(self, cmd: str, vals: Tuple = tuple()) -> List: self._cursor.execute(cmd, vals) return self._cursor.fetchall() |
| Line 25: | Line 25: |
| def execute( self, cmd: str, vals: Tuple = tuple(), ) -> List: self.curs.execute(cmd, vals) return self.curs.fetchone() |
class database_connection(object): def __init__(self, filename: str) -> None: self._connect = sqlite3.connect(filename, detect_types=sqlite3.PARSE_DECLTYPES) self._cursor = self._connect.cursor() def __enter__(self) -> sqlite3.Cursor: return Database(self._cursor) def __exit__(self, typ: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None) -> bool: self._connect.commit() self._cursor.close() self._connect.close() with database_connection(":memory:") as db: db.execute("CREATE TABLE mytable (num INTEGER, name TEXT)") for n in range(10): db.execute("INSERT INTO mytable (num,name) VALUES (?,?)", (n,str(n))) data = db.execute("SELECT * FROM mytable") pprint(data) }}} ---- |
| Line 34: | Line 49: |
| class open_db(object): """Context manager for sqlite3 DB.""" |
|
| Line 37: | Line 50: |
| def __init__( self, filename: str, ) -> None: self.conn = sqlite3.connect(filename, detect_types=sqlite3.PARSE_DECLTYPES) self.curs = self.conn.cursor() |
== Usage == |
| Line 44: | Line 52: |
| def __enter__(self) -> Database: return Database(self.curs) |
A context manager needs to define an `__enter__` function and an `__exit__` function. |
| Line 47: | Line 54: |
| def __exit__( self, *exception: Any, ) -> None: self.conn.commit() self.curs.close() self.conn.close() with open_db("my.sqlite") as db: db.execute("CREATE TABLE mytable (num INTEGER, name TEXT)") db.execute("INSERT INTO mytable (num,name) VALUES (?,?)", (1,2)) data = db.execute("SELECT * FROM mytable") pprint.pprint(data) }}} |
The `__exit__` function is called if an error raises. To suppress errors, have it return `True`. |
Python Context Managers
A context manager is an object that forms a scope inside a with block.
Example
import sqlite3
from pprint import pprint
from types import TracebackType
class Database(object):
def __init__(self, cursor: sqlite3.Cursor) -> None:
self._cursor = cursor
def execute(self, cmd: str, vals: Tuple = tuple()) -> List:
self._cursor.execute(cmd, vals)
return self._cursor.fetchall()
class database_connection(object):
def __init__(self, filename: str) -> None:
self._connect = sqlite3.connect(filename, detect_types=sqlite3.PARSE_DECLTYPES)
self._cursor = self._connect.cursor()
def __enter__(self) -> sqlite3.Cursor:
return Database(self._cursor)
def __exit__(self, typ: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None) -> bool:
self._connect.commit()
self._cursor.close()
self._connect.close()
with database_connection(":memory:") as db:
db.execute("CREATE TABLE mytable (num INTEGER, name TEXT)")
for n in range(10):
db.execute("INSERT INTO mytable (num,name) VALUES (?,?)", (n,str(n)))
data = db.execute("SELECT * FROM mytable")
pprint(data)
Usage
A context manager needs to define an __enter__ function and an __exit__ function.
The __exit__ function is called if an error raises. To suppress errors, have it return True.