Coverage for src/pypermission/db.py: 100%

1from sqlite3 import Connection 

2 

3from sqlalchemy.engine.base import Engine 

4from sqlalchemy.event import contains 

5from sqlalchemy.pool.base import ( 

6 _ConnectionRecord, # pyright: ignore[reportPrivateUsage] 

7) 

8 

9from pypermission.exc import PyPermissionError 

10from pypermission.models import BaseORM 

11 

12 

13def create_rbac_database_table(*, engine: Engine) -> None: 

14 """ 

15 Create all required database tables via. SQLAlchemy for PyPermission. 

16 

17 Parameters 

18 ---------- 

19 engine : Engine 

20 The SQLAlchemy engine. 

21 """ 

22 if engine.driver == "pysqlite" and not contains( 

23 engine, "connect", set_sqlite_pragma 

24 ): 

25 raise PyPermissionError( 

26 "Foreign keys pragma appears to not be set! Please use the 'set_sqlite_pragma' function" 

27 " on your SQLite engine before interacting with the database!" 

28 ) 

29 

30 BaseORM.metadata.create_all(bind=engine) 

31 

32 

33# https://docs.sqlalchemy.org/en/20/dialects/sqlite.html#foreign-key-support 

34def set_sqlite_pragma( 

35 dbapi_connection: Connection, _connection_record: _ConnectionRecord 

36) -> None: 

37 """ 

38 Enable foreign key constraints for SQLite connections. 

39 

40 This function ensures that SQLite foreign key constraints are enabled. 

41 

42 Notes 

43 ----- 

44 SQLite's foreign key support is disabled by default and requires explicit 

45 enabling. The sqlite driver will not set PRAGMA foreign_keys if 

46 autocommit=False, which is why this function temporarily enables it. 

47 

48 References 

49 ---------- 

50 - SQLAlchemy SQLite Foreign Key Support: 

51 https://docs.sqlalchemy.org/en/20/dialects/sqlite.html#foreign-key-support 

52 

53 Examples 

54 -------- 

55 >>> from sqlalchemy import create_engine 

56 >>> from sqlalchemy.event import listen 

57 >>> engine = create_engine('sqlite:///example.db') 

58 >>> listen(engine, 'connect', set_sqlite_pragma) 

59 """ 

60 ac = dbapi_connection.autocommit 

61 dbapi_connection.autocommit = True 

62 

63 cursor = dbapi_connection.cursor() 

64 cursor.execute("PRAGMA foreign_keys=ON") 

65 cursor.close() 

66 

67 dbapi_connection.autocommit = ac