Coverage for python/felis/db/schema.py: 38%

1"""Database utilities for Felis schemas.""" 

2 

3# This file is part of felis. 

4# 

5# Developed for the LSST Data Management System. 

6# This product includes software developed by the LSST Project 

7# (https://www.lsst.org). 

8# See the COPYRIGHT file at the top-level directory of this distribution 

9# for details of code ownership. 

10# 

11# This program is free software: you can redistribute it and/or modify 

12# it under the terms of the GNU General Public License as published by 

13# the Free Software Foundation, either version 3 of the License, or 

14# (at your option) any later version. 

15# 

16# This program is distributed in the hope that it will be useful, 

17# but WITHOUT ANY WARRANTY; without even the implied warranty of 

18# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

19# GNU General Public License for more details. 

20# 

21# You should have received a copy of the GNU General Public License 

22# along with this program. If not, see <https://www.gnu.org/licenses/>. 

23 

24from sqlalchemy import Engine, create_engine 

25 

26from ..datamodel import Schema 

27from ..metadata import MetaDataBuilder 

28from .utils import DatabaseContext 

29 

30__all__ = ["create_database"] 

31 

32 

33def create_database(schema: Schema, engine_or_url_str: Engine | str | None = None) -> DatabaseContext: 

34 """ 

35 Create a database from the specified `Schema`. 

36 

37 Parameters 

38 ---------- 

39 schema 

40 The schema to create. 

41 engine_or_url_str 

42 The SQLAlchemy engine or URL to use for database creation. 

43 If None, an in-memory SQLite database will be created. 

44 

45 Returns 

46 ------- 

47 `DatabaseContext` 

48 The database context object. 

49 """ 

50 if engine_or_url_str is not None: 

51 engine = ( 

52 engine_or_url_str if isinstance(engine_or_url_str, Engine) else create_engine(engine_or_url_str) 

53 ) 

54 else: 

55 engine = create_engine("sqlite:///:memory:") 

56 metadata = MetaDataBuilder( 

57 schema, apply_schema_to_metadata=False if engine.url.drivername == "sqlite" else True 

58 ).build() 

59 ctx = DatabaseContext(metadata, engine) 

60 ctx.initialize() 

61 ctx.create_all() 

62 return ctx