Coverage for src/lexigram/auth/cli/schema_setup.py: 100%

1"""Schema setup contributions for the lexigram-auth package.""" 

2 

3from __future__ import annotations 

4 

5from lexigram.contracts.cli.contributions import SchemaSetupOutcome, SchemaSetupResult 

6from lexigram.contracts.data.sql.database import DatabaseProviderProtocol 

7 

8_OAUTH_TABLE = "oauth_identities" 

9 

10 

11async def ensure_oauth_identities(db: DatabaseProviderProtocol) -> SchemaSetupOutcome: 

12 """Ensure the ``oauth_identities`` table exists. 

13 

14 Args: 

15 db: The database provider. 

16 

17 Returns: 

18 Outcome reporting CREATED, ALREADY_PRESENT, or FAILED. 

19 """ 

20 from lexigram.auth.storage.oauth_identity_store import SQLAlchemyOAuthIdentityStore 

21 

22 try: 

23 existed = await db.table_exists(_OAUTH_TABLE) 

24 await SQLAlchemyOAuthIdentityStore(db)._ensure_tables() 

25 return SchemaSetupOutcome( 

26 status=( 

27 SchemaSetupResult.ALREADY_PRESENT 

28 if existed 

29 else SchemaSetupResult.CREATED 

30 ) 

31 ) 

32 except Exception as exc: 

33 return SchemaSetupOutcome(status=SchemaSetupResult.FAILED, message=str(exc))