Coverage for python/lsst/daf/butler/registry/interfaces/_database.py: 90%
466 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-10 08:54 +0000
1# This file is part of daf_butler.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <http://www.gnu.org/licenses/>.
27from __future__ import annotations
29from ... import ddl, time_utils
31__all__ = [
32 "Database",
33 "DatabaseConflictError",
34 "DatabaseInsertMode",
35 "DatabaseMetadata",
36 "ReadOnlyDatabaseError",
37 "SchemaAlreadyDefinedError",
38 "StaticTablesContext",
39]
41import enum
42import os
43import sys
44import uuid
45import warnings
46from abc import ABC, abstractmethod
47from collections import defaultdict
48from collections.abc import Callable, Iterable, Iterator, Sequence
49from contextlib import contextmanager
50from threading import Lock
51from typing import Any, TypeVar, cast, final
53import astropy.time
54import sqlalchemy
56from ..._named import NamedValueAbstractSet
57from ...name_shrinker import NameShrinker
58from ...timespan_database_representation import TimespanDatabaseRepresentation
59from .._exceptions import ConflictingDefinitionError
60from ._database_explain import get_query_plan
62_T = TypeVar("_T")
65class DatabaseInsertMode(enum.Enum):
66 """Mode options available for inserting database records."""
68 INSERT = enum.auto()
69 """Insert records, failing if they already exist."""
71 REPLACE = enum.auto()
72 """Replace records, overwriting existing."""
74 ENSURE = enum.auto()
75 """Insert records, skipping any that already exist."""
78# TODO: method is called with list[ReflectedColumn] in SA 2, and
79# ReflectedColumn does not exist in 1.4.
80def _checkExistingTableDefinition(name: str, spec: ddl.TableSpec, inspection: list) -> None:
81 """Test that the definition of a table in a `ddl.TableSpec` and from
82 database introspection are consistent.
84 Parameters
85 ----------
86 name : `str`
87 Name of the table (only used in error messages).
88 spec : `ddl.TableSpec`
89 Specification of the table.
90 inspection : `dict`
91 Dictionary returned by
92 `sqlalchemy.engine.reflection.Inspector.get_columns`.
94 Raises
95 ------
96 DatabaseConflictError
97 Raised if the definitions are inconsistent.
98 """
99 columnNames = [c["name"] for c in inspection]
100 if spec.fields.names != set(columnNames): 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true
101 raise DatabaseConflictError(
102 f"Table '{name}' exists but is defined differently in the database; "
103 f"specification has columns {list(spec.fields.names)}, while the "
104 f"table in the database has {columnNames}."
105 )
108class ReadOnlyDatabaseError(RuntimeError):
109 """Exception raised when a write operation is called on a read-only
110 `Database`.
111 """
114class DatabaseConflictError(ConflictingDefinitionError):
115 """Exception raised when database content (row values or schema entities)
116 are inconsistent with what this client expects.
117 """
120class SchemaAlreadyDefinedError(RuntimeError):
121 """Exception raised when trying to initialize database schema when some
122 tables already exist.
123 """
126class StaticTablesContext:
127 """Helper class used to declare the static schema for a registry layer
128 in a database.
130 An instance of this class is returned by `Database.declareStaticTables`,
131 which should be the only way it should be constructed.
133 Parameters
134 ----------
135 db : `Database`
136 The database.
137 connection : `sqlalchemy.engine.Connection`
138 The connection object.
139 """
141 def __init__(self, db: Database, connection: sqlalchemy.engine.Connection):
142 self._db = db
143 self._inspector = sqlalchemy.inspect(connection)
144 self._tableNames = frozenset(self._inspector.get_table_names(schema=self._db.namespace))
145 self._initializers: list[Callable[[Database], None]] = []
147 def addTable(self, name: str, spec: ddl.TableSpec) -> sqlalchemy.schema.Table:
148 """Add a new table to the schema, returning its sqlalchemy
149 representation.
151 Parameters
152 ----------
153 name : `str`
154 The name of the table.
155 spec : `ddl.TableSpec`
156 The specification of the table.
158 Returns
159 -------
160 table : `sqlalchemy.schema.Table`
161 The created table.
163 Notes
164 -----
165 The new table may not actually be created until the end of the
166 context created by `Database.declareStaticTables`, allowing tables
167 to be declared in any order even in the presence of foreign key
168 relationships.
169 """
170 metadata = self._db._metadata
171 assert metadata is not None, "Guaranteed by context manager that returns this object."
172 return metadata.add_table(self._db, name, spec)
174 def addTableTuple(self, specs: tuple[ddl.TableSpec, ...]) -> tuple[sqlalchemy.schema.Table, ...]:
175 """Add a named tuple of tables to the schema, returning their
176 SQLAlchemy representations in a named tuple of the same type.
178 The new tables may not actually be created until the end of the
179 context created by `Database.declareStaticTables`, allowing tables
180 to be declared in any order even in the presence of foreign key
181 relationships.
183 Parameters
184 ----------
185 specs : `tuple` of `ddl.TableSpec`
186 Specifications of multiple tables.
188 Returns
189 -------
190 tables : `tuple` of `sqlalchemy.schema.Table`
191 All the tables created.
193 Notes
194 -----
195 ``specs`` *must* be an instance of a type created by
196 `collections.namedtuple`, not just regular tuple, and the returned
197 object is guaranteed to be the same. Because `~collections.namedtuple`
198 is just a factory for `type` objects, not an actual type itself,
199 we cannot represent this with type annotations.
200 """
201 return specs._make( # type: ignore
202 self.addTable(name, spec)
203 for name, spec in zip(specs._fields, specs, strict=True) # type: ignore
204 )
206 def addInitializer(self, initializer: Callable[[Database], None]) -> None:
207 """Add a method that does one-time initialization of a database.
209 Initialization can mean anything that changes state of a database
210 and needs to be done exactly once after database schema was created.
211 An example for that could be population of schema attributes.
213 Parameters
214 ----------
215 initializer : `~collections.abc.Callable`
216 Method of a single argument which is a `Database` instance.
217 """
218 self._initializers.append(initializer)
221class Database(ABC):
222 """An abstract interface that represents a particular database engine's
223 representation of a single schema/namespace/database.
225 Parameters
226 ----------
227 origin : `int`
228 An integer ID that should be used as the default for any datasets,
229 quanta, or other entities that use a (autoincrement, origin) compound
230 primary key.
231 engine : `sqlalchemy.engine.Engine`
232 The SQLAlchemy engine for this `Database`.
233 namespace : `str`, optional
234 Name of the schema or namespace this instance is associated with.
235 This is passed as the ``schema`` argument when constructing a
236 `sqlalchemy.schema.MetaData` instance. We use ``namespace`` instead to
237 avoid confusion between "schema means namespace" and "schema means
238 table definitions".
239 metadata : `sqlalchemy.schema.MetaData`, optional
240 Object representing the tables and other schema entities. If not
241 provided, will be generated during the next call to
242 ``declareStaticTables``.
243 allow_temporary_tables : `bool`, optional
244 If `True`, database operations will be allowed to use temporary tables.
245 If `False`, other SQL constructs will be used instead of temporary
246 tables when possible.
248 Notes
249 -----
250 `Database` requires all write operations to go through its special named
251 methods. Our write patterns are sufficiently simple that we don't really
252 need the full flexibility of SQL insert/update/delete syntax, and we need
253 non-standard (but common) functionality in these operations sufficiently
254 often that it seems worthwhile to provide our own generic API.
256 In contrast, `Database.query` allows arbitrary ``SELECT`` queries (via
257 their SQLAlchemy representation) to be run, as we expect these to require
258 significantly more sophistication while still being limited to standard
259 SQL.
261 `Database` itself has several underscore-prefixed attributes:
263 - ``_engine``: SQLAlchemy object representing its engine.
264 - ``_connection``: method returning a context manager for
265 `sqlalchemy.engine.Connection` object.
266 - ``_metadata``: the `sqlalchemy.schema.MetaData` object representing
267 the tables and other schema entities.
269 These are considered protected (derived classes may access them, but other
270 code should not), and read-only, aside from executing SQL via
271 ``_connection``.
272 """
274 def __init__(
275 self,
276 *,
277 origin: int,
278 engine: sqlalchemy.engine.Engine,
279 namespace: str | None = None,
280 metadata: DatabaseMetadata | None = None,
281 allow_temporary_tables: bool = True,
282 ):
283 self.origin = origin
284 self.name_shrinker = NameShrinker(engine.dialect.max_identifier_length)
285 self.namespace = namespace
286 self._engine = engine
287 self._session_connection: sqlalchemy.engine.Connection | None = None
288 self._temp_tables: set[str] = set()
289 self._metadata = metadata
290 self._allow_temporary_tables = allow_temporary_tables
292 def dispose(self) -> None:
293 """Close all open database connections held by this `Database`
294 instance.
295 """
296 self._engine.dispose()
298 def __repr__(self) -> str:
299 # Rather than try to reproduce all the parameters used to create
300 # the object, instead report the more useful information of the
301 # connection URL.
302 if self._engine.url.password is not None: 302 ↛ 303line 302 didn't jump to line 303 because the condition on line 302 was never true
303 uri = str(self._engine.url.set(password="***"))
304 else:
305 uri = str(self._engine.url)
306 if self.namespace: 306 ↛ 307line 306 didn't jump to line 307 because the condition on line 306 was never true
307 uri += f"#{self.namespace}"
308 return f'{type(self).__name__}("{uri}")'
310 @classmethod
311 def makeDefaultUri(cls, root: str) -> str | None:
312 """Create a default connection URI appropriate for the given root
313 directory, or `None` if there can be no such default.
315 Parameters
316 ----------
317 root : `str`
318 Root string to use to build connection URI.
320 Returns
321 -------
322 uri : `str` or `None`
323 The URI string or `None`.
324 """
325 return None
327 @classmethod
328 def fromUri(
329 cls,
330 uri: str | sqlalchemy.engine.URL,
331 *,
332 origin: int,
333 namespace: str | None = None,
334 writeable: bool = True,
335 allow_temporary_tables: bool = True,
336 ) -> Database:
337 """Construct a database from a SQLAlchemy URI.
339 Parameters
340 ----------
341 uri : `str` or `sqlalchemy.engine.URL`
342 A SQLAlchemy URI connection string.
343 origin : `int`
344 An integer ID that should be used as the default for any datasets,
345 quanta, or other entities that use a (autoincrement, origin)
346 compound primary key.
347 namespace : `str`, optional
348 A database namespace (i.e. schema) the new instance should be
349 associated with. If `None` (default), the namespace (if any) is
350 inferred from the URI.
351 writeable : `bool`, optional
352 If `True`, allow write operations on the database, including
353 ``CREATE TABLE``.
354 allow_temporary_tables : `bool`, optional
355 If `True`, database operations will be allowed to use temporary
356 tables.
357 If `False`, other SQL constructs will be used instead of temporary
358 tables when possible.
360 Returns
361 -------
362 db : `Database`
363 A new `Database` instance.
364 """
365 db = cls.fromEngine(
366 cls.makeEngine(uri, writeable=writeable), origin=origin, namespace=namespace, writeable=writeable
367 )
368 db._allow_temporary_tables = allow_temporary_tables
369 return db
371 @abstractmethod
372 def clone(self) -> Database:
373 """Make an independent copy of this `Database` object.
375 Returns
376 -------
377 db : `Database`
378 A new `Database` instance with the same configuration as this
379 instance.
380 """
381 raise NotImplementedError()
383 @classmethod
384 @abstractmethod
385 def makeEngine(
386 cls, uri: str | sqlalchemy.engine.URL, *, writeable: bool = True
387 ) -> sqlalchemy.engine.Engine:
388 """Create a `sqlalchemy.engine.Engine` from a SQLAlchemy URI.
390 Parameters
391 ----------
392 uri : `str` or `sqlalchemy.engine.URL`
393 A SQLAlchemy URI connection string.
394 writeable : `bool`, optional
395 If `True`, allow write operations on the database, including
396 ``CREATE TABLE``.
398 Returns
399 -------
400 engine : `sqlalchemy.engine.Engine`
401 A database engine.
403 Notes
404 -----
405 Subclasses that support other ways to connect to a database are
406 encouraged to add optional arguments to their implementation of this
407 method, as long as they maintain compatibility with the base class
408 call signature.
409 """
410 raise NotImplementedError()
412 @classmethod
413 @abstractmethod
414 def fromEngine(
415 cls,
416 engine: sqlalchemy.engine.Engine,
417 *,
418 origin: int,
419 namespace: str | None = None,
420 writeable: bool = True,
421 ) -> Database:
422 """Create a new `Database` from an existing `sqlalchemy.engine.Engine`.
424 Parameters
425 ----------
426 engine : `sqlalchemy.engine.Engine`
427 The engine for the database. May be shared between `Database`
428 instances.
429 origin : `int`
430 An integer ID that should be used as the default for any datasets,
431 quanta, or other entities that use a (autoincrement, origin)
432 compound primary key.
433 namespace : `str`, optional
434 A different database namespace (i.e. schema) the new instance
435 should be associated with. If `None` (default), the namespace
436 (if any) is inferred from the connection.
437 writeable : `bool`, optional
438 If `True`, allow write operations on the database, including
439 ``CREATE TABLE``.
441 Returns
442 -------
443 db : `Database`
444 A new `Database` instance.
446 Notes
447 -----
448 This method allows different `Database` instances to share the same
449 engine, which is desirable when they represent different namespaces
450 can be queried together.
451 """
452 raise NotImplementedError()
454 @final
455 @contextmanager
456 def session(self) -> Iterator[None]:
457 """Return a context manager that represents a session (persistent
458 connection to a database).
460 Returns
461 -------
462 context : `AbstractContextManager` [ `None` ]
463 A context manager that does not return a value when entered.
465 Notes
466 -----
467 This method should be used when a sequence of read-only SQL operations
468 will be performed in rapid succession *without* a requirement that they
469 yield consistent results in the presence of concurrent writes (or, more
470 rarely, when conflicting concurrent writes are rare/impossible and the
471 session will be open long enough that a transaction is inadvisable).
472 """
473 with self._session():
474 yield
476 @final
477 @contextmanager
478 def transaction(
479 self,
480 *,
481 interrupting: bool = False,
482 savepoint: bool = False,
483 lock: Iterable[sqlalchemy.schema.Table] = (),
484 for_temp_tables: bool = False,
485 ) -> Iterator[None]:
486 """Return a context manager that represents a transaction.
488 Parameters
489 ----------
490 interrupting : `bool`, optional
491 If `True` (`False` is default), this transaction block may not be
492 nested within an outer one, and attempting to do so is a logic
493 (i.e. assertion) error.
494 savepoint : `bool`, optional
495 If `True` (`False` is default), create a ``SAVEPOINT``, allowing
496 exceptions raised by the database (e.g. due to constraint
497 violations) during this transaction's context to be caught outside
498 it without also rolling back all operations in an outer transaction
499 block. If `False`, transactions may still be nested, but a
500 rollback may be generated at any level and affects all levels, and
501 commits are deferred until the outermost block completes. If any
502 outer transaction block was created with ``savepoint=True``, all
503 inner blocks will be as well (regardless of the actual value
504 passed). This has no effect if this is the outermost transaction.
505 lock : `~collections.abc.Iterable` [ `sqlalchemy.schema.Table` ], \
506 optional
507 A list of tables to lock for the duration of this transaction.
508 These locks are guaranteed to prevent concurrent writes and allow
509 this transaction (only) to acquire the same locks (others should
510 block), but only prevent concurrent reads if the database engine
511 requires that in order to block concurrent writes.
512 for_temp_tables : `bool`, optional
513 If `True`, this transaction may involve creating temporary tables.
515 Returns
516 -------
517 context : `AbstractContextManager` [ `None` ]
518 A context manager that commits the transaction when it is exited
519 without error and rolls back the transactoin when it is exited via
520 an exception.
522 Notes
523 -----
524 All transactions on a connection managed by one or more `Database`
525 instances _must_ go through this method, or transaction state will not
526 be correctly managed.
527 """
528 with self._transaction(
529 interrupting=interrupting, savepoint=savepoint, lock=lock, for_temp_tables=for_temp_tables
530 ):
531 yield
533 @contextmanager
534 def temporary_table(
535 self, spec: ddl.TableSpec, name: str | None = None
536 ) -> Iterator[sqlalchemy.schema.Table]:
537 """Return a context manager that creates and then drops a temporary
538 table.
540 Parameters
541 ----------
542 spec : `ddl.TableSpec`
543 Specification for the columns. Unique and foreign key constraints
544 may be ignored.
545 name : `str`, optional
546 If provided, the name of the SQL construct. If not provided, an
547 opaque but unique identifier is generated.
549 Returns
550 -------
551 context : `AbstractContextManager` [ `sqlalchemy.schema.Table` ]
552 A context manager that returns a SQLAlchemy representation of the
553 temporary table when entered.
555 Notes
556 -----
557 Temporary tables may be created, dropped, and written to even in
558 read-only databases - at least according to the Python-level
559 protections in the `Database` classes. Server permissions may say
560 otherwise, but in that case they probably need to be modified to
561 support the full range of expected read-only butler behavior.
562 """
563 assert self._metadata is not None, "Static tables must be created before temporary tables"
564 if not self.supports_temporary_tables:
565 raise ReadOnlyDatabaseError("Creation of temporary tables is not supported by this database.")
566 with self._session() as connection:
567 table = self._make_temporary_table(connection, spec=spec, name=name)
568 self._temp_tables.add(table.key)
569 try:
570 yield table
571 finally:
572 self._temp_tables.remove(table.key)
573 self._metadata.remove_table(table.name)
575 @contextmanager
576 def _session(self) -> Iterator[sqlalchemy.engine.Connection]:
577 """Protected implementation for `session` that actually returns the
578 connection.
580 This method is for internal `Database` calls that need the actual
581 SQLAlchemy connection object. It should be overridden by subclasses
582 instead of `session` itself.
584 Returns
585 -------
586 context : `AbstractContextManager` [ `sqlalchemy.engine.Connection` ]
587 A context manager that returns a SQLALchemy connection when
588 entered.
590 """
591 if self._session_connection is not None:
592 # session already started, just reuse that
593 yield self._session_connection
594 else:
595 try:
596 # open new connection and close it when done
597 self._session_connection = self._engine.connect()
598 yield self._session_connection
599 finally:
600 if self._session_connection is not None: 600 ↛ 604line 600 didn't jump to line 604 because the condition on line 600 was always true
601 self._session_connection.close()
602 self._session_connection = None
603 # Temporary tables only live within session
604 self._temp_tables = set()
606 @contextmanager
607 def _transaction(
608 self,
609 *,
610 interrupting: bool = False,
611 savepoint: bool = False,
612 lock: Iterable[sqlalchemy.schema.Table] = (),
613 for_temp_tables: bool = False,
614 ) -> Iterator[tuple[bool, sqlalchemy.engine.Connection]]:
615 """Protected implementation for `transaction` that actually returns the
616 connection and whether this is a new outermost transaction.
618 This method is for internal `Database` calls that need the actual
619 SQLAlchemy connection object. It should be overridden by subclasses
620 instead of `transaction` itself.
622 Parameters
623 ----------
624 interrupting : `bool`, optional
625 If `True` (`False` is default), this transaction block may not be
626 nested without an outer one, and attempting to do so is a logic
627 (i.e. assertion) error.
628 savepoint : `bool`, optional
629 If `True` (`False` is default), create a ``SAVEPOINT``, allowing
630 exceptions raised by the database (e.g. due to constraint
631 violations) during this transaction's context to be caught outside
632 it without also rolling back all operations in an outer transaction
633 block. If `False`, transactions may still be nested, but a
634 rollback may be generated at any level and affects all levels, and
635 commits are deferred until the outermost block completes. If any
636 outer transaction block was created with ``savepoint=True``, all
637 inner blocks will be as well (regardless of the actual value
638 passed). This has no effect if this is the outermost transaction.
639 lock : `~collections.abc.Iterable` [ `sqlalchemy.schema.Table` ], \
640 optional
641 A list of tables to lock for the duration of this transaction.
642 These locks are guaranteed to prevent concurrent writes and allow
643 this transaction (only) to acquire the same locks (others should
644 block), but only prevent concurrent reads if the database engine
645 requires that in order to block concurrent writes.
646 for_temp_tables : `bool`, optional
647 If `True`, this transaction may involve creating temporary tables.
649 Returns
650 -------
651 context : `AbstractContextManager` [ `tuple` [ `bool`,
652 `sqlalchemy.engine.Connection` ] ]
653 A context manager that commits the transaction when it is exited
654 without error and rolls back the transactoin when it is exited via
655 an exception. When entered, it returns a tuple of:
657 - ``is_new`` (`bool`): whether this is a new (outermost)
658 transaction;
659 - ``connection`` (`sqlalchemy.engine.Connection`): the connection.
660 """
661 with self._session() as connection:
662 already_in_transaction = self.isInTransaction()
663 assert not (interrupting and already_in_transaction), (
664 "Logic error in transaction nesting: an operation that would "
665 "interrupt the active transaction context has been requested."
666 )
667 savepoint = savepoint or connection.in_nested_transaction()
668 trans: sqlalchemy.engine.Transaction | None
669 if already_in_transaction:
670 if savepoint:
671 trans = connection.begin_nested()
672 else:
673 # Nested non-savepoint transactions don't do anything.
674 trans = None
675 else:
676 # Use a regular (non-savepoint) transaction always for the
677 # outermost context.
678 trans = connection.begin()
679 self._lockTables(connection, lock)
680 try:
681 yield not already_in_transaction, connection
682 if trans is not None:
683 trans.commit()
684 except BaseException:
685 if trans is not None:
686 trans.rollback()
687 raise
689 @abstractmethod
690 def _lockTables(
691 self, connection: sqlalchemy.engine.Connection, tables: Iterable[sqlalchemy.schema.Table] = ()
692 ) -> None:
693 """Acquire locks on the given tables.
695 This is an implementation hook for subclasses, called by `transaction`.
696 It should not be called directly by other code.
698 Parameters
699 ----------
700 connection : `sqlalchemy.engine.Connection`
701 Database connection object. It is guaranteed that transaction is
702 already in a progress for this connection.
703 tables : `~collections.abc.Iterable` [ `sqlalchemy.schema.Table` ], \
704 optional
705 A list of tables to lock for the duration of this transaction.
706 These locks are guaranteed to prevent concurrent writes and allow
707 this transaction (only) to acquire the same locks (others should
708 block), but only prevent concurrent reads if the database engine
709 requires that in order to block concurrent writes.
710 """
711 raise NotImplementedError()
713 def isTableWriteable(self, table: sqlalchemy.schema.Table) -> bool:
714 """Check whether a table is writeable, either because the database
715 connection is read-write or the table is a temporary table.
717 Parameters
718 ----------
719 table : `sqlalchemy.schema.Table`
720 SQLAlchemy table object to check.
722 Returns
723 -------
724 writeable : `bool`
725 Whether this table is writeable.
726 """
727 return self.isWriteable() or table.key in self._temp_tables
729 def assertTableWriteable(self, table: sqlalchemy.schema.Table, msg: str) -> None:
730 """Raise if the given table is not writeable, either because the
731 database connection is read-write or the table is a temporary table.
733 Parameters
734 ----------
735 table : `sqlalchemy.schema.Table`
736 SQLAlchemy table object to check.
737 msg : `str`, optional
738 If provided, raise `ReadOnlyDatabaseError` instead of returning
739 `False`, with this message.
740 """
741 if not self.isTableWriteable(table): 741 ↛ 742line 741 didn't jump to line 742 because the condition on line 741 was never true
742 raise ReadOnlyDatabaseError(msg)
744 @contextmanager
745 def declareStaticTables(self, *, create: bool) -> Iterator[StaticTablesContext]:
746 """Return a context manager in which the database's static DDL schema
747 can be declared.
749 Parameters
750 ----------
751 create : `bool`
752 If `True`, attempt to create all tables at the end of the context.
753 If `False`, they will be assumed to already exist.
755 Returns
756 -------
757 schema : `StaticTablesContext`
758 A helper object that is used to add new tables.
760 Raises
761 ------
762 ReadOnlyDatabaseError
763 Raised if ``create`` is `True`, `Database.isWriteable` is `False`,
764 and one or more declared tables do not already exist.
766 Examples
767 --------
768 Given a `Database` instance ``db``::
770 with db.declareStaticTables(create=True) as schema:
771 schema.addTable("table1", TableSpec(...))
772 schema.addTable("table2", TableSpec(...))
774 Notes
775 -----
776 A database's static DDL schema must be declared before any dynamic
777 tables are managed via calls to `ensureTableExists` or
778 `getExistingTable`. The order in which static schema tables are added
779 inside the context block is unimportant; they will automatically be
780 sorted and added in an order consistent with their foreign key
781 relationships.
782 """
783 if create and not self.isWriteable():
784 raise ReadOnlyDatabaseError(f"Cannot create tables in read-only database {self}.")
786 self._metadata = DatabaseMetadata(self.namespace)
787 try:
788 with self._transaction() as (_, connection):
789 context = StaticTablesContext(self, connection)
790 if create and context._tableNames:
791 # Looks like database is already initalized, to avoid
792 # danger of modifying/destroying valid schema we refuse to
793 # do anything in this case
794 raise SchemaAlreadyDefinedError(f"Cannot create tables in non-empty database {self}.")
795 yield context
796 if create:
797 if ( 797 ↛ 801line 797 didn't jump to line 801 because the condition on line 797 was never true
798 self.namespace is not None
799 and self.namespace not in context._inspector.get_schema_names()
800 ):
801 connection.execute(sqlalchemy.schema.CreateSchema(self.namespace))
802 # In our tables we have columns that make use of sqlalchemy
803 # Sequence objects. There is currently a bug in sqlalchemy
804 # that causes a deprecation warning to be thrown on a
805 # property of the Sequence object when the repr for the
806 # sequence is created. Here a filter is used to catch these
807 # deprecation warnings when tables are created.
808 with warnings.catch_warnings():
809 warnings.simplefilter("ignore", category=sqlalchemy.exc.SADeprecationWarning)
810 self._metadata.create_all(connection)
811 # call all initializer methods sequentially
812 for init in context._initializers: 812 ↛ 813line 812 didn't jump to line 813 because the loop on line 812 never started
813 init(self)
814 except BaseException:
815 self._metadata = None
816 raise
818 @abstractmethod
819 def isWriteable(self) -> bool:
820 """Return `True` if this database can be modified by this client."""
821 raise NotImplementedError()
823 def isInTransaction(self) -> bool:
824 """Return `True` if there is currently a database connection open with
825 an active transaction; `False` otherwise.
826 """
827 session = self._session_connection
828 return session is not None and session.in_transaction()
830 @abstractmethod
831 def __str__(self) -> str:
832 """Return a human-readable identifier for this `Database`, including
833 any namespace or schema that identifies its names within a `Registry`.
834 """
835 raise NotImplementedError()
837 @property
838 def dialect(self) -> sqlalchemy.engine.Dialect:
839 """The SQLAlchemy dialect for this database engine
840 (`sqlalchemy.engine.Dialect`).
841 """
842 return self._engine.dialect
844 def shrinkDatabaseEntityName(self, original: str) -> str:
845 """Return a version of the given name that fits within this database
846 engine's length limits for table, constraint, indexes, and sequence
847 names.
849 Implementations should not assume that simple truncation is safe,
850 because multiple long names often begin with the same prefix.
852 The default implementation simply returns the given name.
854 Parameters
855 ----------
856 original : `str`
857 The original name.
859 Returns
860 -------
861 shrunk : `str`
862 The new, possibly shortened name.
863 """
864 return original
866 def expandDatabaseEntityName(self, shrunk: str) -> str:
867 """Retrieve the original name for a database entity that was too long
868 to fit within the database engine's limits.
870 Parameters
871 ----------
872 shrunk : `str`
873 The original name.
875 Returns
876 -------
877 shrunk : `str`
878 The new, possibly shortened name.
879 """
880 return shrunk
882 def _makeColumnConstraints(self, table: str, spec: ddl.FieldSpec) -> list[sqlalchemy.CheckConstraint]:
883 """Create constraints based on this spec.
885 Parameters
886 ----------
887 table : `str`
888 Name of the table this column is being added to.
889 spec : `FieldSpec`
890 Specification for the field to be added.
892 Returns
893 -------
894 constraint : `list` of `sqlalchemy.CheckConstraint`
895 Constraint added for this column.
896 """
897 # By default we return no additional constraints
898 return []
900 def _convertFieldSpec(
901 self, table: str, spec: ddl.FieldSpec, metadata: sqlalchemy.MetaData, **kwargs: Any
902 ) -> sqlalchemy.schema.Column:
903 """Convert a `FieldSpec` to a `sqlalchemy.schema.Column`.
905 Parameters
906 ----------
907 table : `str`
908 Name of the table this column is being added to.
909 spec : `FieldSpec`
910 Specification for the field to be added.
911 metadata : `sqlalchemy.MetaData`
912 SQLAlchemy representation of the DDL schema this field's table is
913 being added to.
914 **kwargs
915 Additional keyword arguments to forward to the
916 `sqlalchemy.schema.Column` constructor. This is provided to make
917 it easier for derived classes to delegate to ``super()`` while
918 making only minor changes.
920 Returns
921 -------
922 column : `sqlalchemy.schema.Column`
923 SQLAlchemy representation of the field.
924 """
925 args = []
926 if spec.autoincrement:
927 # Generate a sequence to use for auto incrementing for databases
928 # that do not support it natively. This will be ignored by
929 # sqlalchemy for databases that do support it.
930 args.append(
931 sqlalchemy.Sequence(
932 self.shrinkDatabaseEntityName(f"{table}_seq_{spec.name}"), metadata=metadata
933 )
934 )
935 assert spec.doc is None or isinstance(spec.doc, str), f"Bad doc for {table}.{spec.name}."
936 return sqlalchemy.schema.Column(
937 spec.name,
938 spec.getSizedColumnType(),
939 *args,
940 nullable=spec.nullable,
941 primary_key=spec.primaryKey,
942 comment=spec.doc,
943 server_default=spec.default,
944 **kwargs,
945 )
947 def _convertForeignKeySpec(
948 self, table: str, spec: ddl.ForeignKeySpec, metadata: sqlalchemy.MetaData, **kwargs: Any
949 ) -> sqlalchemy.schema.ForeignKeyConstraint:
950 """Convert a `ForeignKeySpec` to a
951 `sqlalchemy.schema.ForeignKeyConstraint`.
953 Parameters
954 ----------
955 table : `str`
956 Name of the table this foreign key is being added to.
957 spec : `ForeignKeySpec`
958 Specification for the foreign key to be added.
959 metadata : `sqlalchemy.MetaData`
960 SQLAlchemy representation of the DDL schema this constraint is
961 being added to.
962 **kwargs
963 Additional keyword arguments to forward to the
964 `sqlalchemy.schema.ForeignKeyConstraint` constructor. This is
965 provided to make it easier for derived classes to delegate to
966 ``super()`` while making only minor changes.
968 Returns
969 -------
970 constraint : `sqlalchemy.schema.ForeignKeyConstraint`
971 SQLAlchemy representation of the constraint.
972 """
973 name = self.shrinkDatabaseEntityName(
974 "_".join(["fkey", table, spec.table] + list(spec.target) + list(spec.source))
975 )
976 return sqlalchemy.schema.ForeignKeyConstraint(
977 spec.source,
978 [f"{spec.table}.{col}" for col in spec.target],
979 name=name,
980 ondelete=spec.onDelete,
981 )
983 def _convertExclusionConstraintSpec(
984 self,
985 table: str,
986 spec: tuple[str | type[TimespanDatabaseRepresentation], ...],
987 metadata: sqlalchemy.MetaData,
988 ) -> sqlalchemy.schema.Constraint:
989 """Convert a `tuple` from `ddl.TableSpec.exclusion` into a SQLAlchemy
990 constraint representation.
992 Parameters
993 ----------
994 table : `str`
995 Name of the table this constraint is being added to.
996 spec : `tuple` [ `str` or `type` ]
997 A tuple of `str` column names and the `type` object returned by
998 `getTimespanRepresentation` (which must appear exactly once),
999 indicating the order of the columns in the index used to back the
1000 constraint.
1001 metadata : `sqlalchemy.MetaData`
1002 SQLAlchemy representation of the DDL schema this constraint is
1003 being added to.
1005 Returns
1006 -------
1007 constraint : `sqlalchemy.schema.Constraint`
1008 SQLAlchemy representation of the constraint.
1010 Raises
1011 ------
1012 NotImplementedError
1013 Raised if this database does not support exclusion constraints.
1014 """
1015 raise NotImplementedError(f"Database {self} does not support exclusion constraints.")
1017 def _convertTableSpec(
1018 self, name: str, spec: ddl.TableSpec, metadata: sqlalchemy.MetaData, **kwargs: Any
1019 ) -> sqlalchemy.schema.Table:
1020 """Convert a `TableSpec` to a `sqlalchemy.schema.Table`.
1022 Parameters
1023 ----------
1024 name : `str`
1025 The name of the table.
1026 spec : `TableSpec`
1027 Specification for the foreign key to be added.
1028 metadata : `sqlalchemy.MetaData`
1029 SQLAlchemy representation of the DDL schema this table is being
1030 added to.
1031 **kwargs
1032 Additional keyword arguments to forward to the
1033 `sqlalchemy.schema.Table` constructor. This is provided to make it
1034 easier for derived classes to delegate to ``super()`` while making
1035 only minor changes.
1037 Returns
1038 -------
1039 table : `sqlalchemy.schema.Table`
1040 SQLAlchemy representation of the table.
1042 Notes
1043 -----
1044 This method does not handle ``spec.foreignKeys`` at all, in order to
1045 avoid circular dependencies. These are added by higher-level logic in
1046 `ensureTableExists`, `getExistingTable`, and `declareStaticTables`.
1047 """
1048 args: list[sqlalchemy.schema.SchemaItem] = [
1049 self._convertFieldSpec(name, fieldSpec, metadata) for fieldSpec in spec.fields
1050 ]
1052 # Add any column constraints
1053 for fieldSpec in spec.fields:
1054 args.extend(self._makeColumnConstraints(name, fieldSpec))
1056 # Track indexes added for primary key and unique constraints, to make
1057 # sure we don't add duplicate explicit or foreign key indexes for
1058 # those.
1059 allIndexes = {tuple(fieldSpec.name for fieldSpec in spec.fields if fieldSpec.primaryKey)}
1060 args.extend(
1061 sqlalchemy.schema.UniqueConstraint(
1062 *columns, name=self.shrinkDatabaseEntityName("_".join([name, "unq"] + list(columns)))
1063 )
1064 for columns in spec.unique
1065 )
1066 allIndexes.update(spec.unique)
1067 args.extend(
1068 sqlalchemy.schema.Index(
1069 self.shrinkDatabaseEntityName("_".join([name, "idx"] + list(index.columns))),
1070 *index.columns,
1071 unique=(index.columns in spec.unique),
1072 **index.kwargs,
1073 )
1074 for index in spec.indexes
1075 if index.columns not in allIndexes
1076 )
1077 allIndexes.update(index.columns for index in spec.indexes)
1078 args.extend(
1079 sqlalchemy.schema.Index(
1080 self.shrinkDatabaseEntityName("_".join((name, "fkidx") + fk.source)),
1081 *fk.source,
1082 )
1083 for fk in spec.foreignKeys
1084 if fk.addIndex and fk.source not in allIndexes
1085 )
1087 args.extend(self._convertExclusionConstraintSpec(name, excl, metadata) for excl in spec.exclusion)
1089 assert spec.doc is None or isinstance(spec.doc, str), f"Bad doc for {name}."
1090 return sqlalchemy.schema.Table(name, metadata, *args, comment=spec.doc, info={"spec": spec}, **kwargs)
1092 def ensureTableExists(self, name: str, spec: ddl.TableSpec) -> sqlalchemy.schema.Table:
1093 """Ensure that a table with the given name and specification exists,
1094 creating it if necessary.
1096 Parameters
1097 ----------
1098 name : `str`
1099 Name of the table (not including namespace qualifiers).
1100 spec : `TableSpec`
1101 Specification for the table. This will be used when creating the
1102 table, and *may* be used when obtaining an existing table to check
1103 for consistency, but no such check is guaranteed.
1105 Returns
1106 -------
1107 table : `sqlalchemy.schema.Table`
1108 SQLAlchemy representation of the table.
1110 Raises
1111 ------
1112 ReadOnlyDatabaseError
1113 Raised if `isWriteable` returns `False`, and the table does not
1114 already exist.
1115 DatabaseConflictError
1116 Raised if the table exists but ``spec`` is inconsistent with its
1117 definition.
1119 Notes
1120 -----
1121 This method may not be called within transactions. It may be called on
1122 read-only databases if and only if the table does in fact already
1123 exist.
1125 Subclasses may override this method, but usually should not need to.
1126 """
1127 # TODO: if _engine is used to make a table then it uses separate
1128 # connection and should not interfere with current transaction
1129 assert not self.isInTransaction(), "Table creation interrupts transactions."
1130 assert self._metadata is not None, "Static tables must be declared before dynamic tables."
1131 table = self.getExistingTable(name, spec)
1132 if table is not None:
1133 return table
1134 if not self.isWriteable():
1135 raise ReadOnlyDatabaseError(
1136 f"Table {name} does not exist, and cannot be created because database {self} is read-only."
1137 )
1139 table = self._metadata.add_table(self, name, spec)
1140 try:
1141 with self._transaction() as (_, connection):
1142 table.create(connection)
1143 except sqlalchemy.exc.DatabaseError:
1144 # Some other process could have created the table meanwhile, which
1145 # usually causes OperationalError or ProgrammingError. We cannot
1146 # use IF NOT EXISTS clause in this case due to PostgreSQL race
1147 # condition on server side which causes IntegrityError. Instead we
1148 # catch these exceptions (they all inherit DatabaseError) and
1149 # re-check whether table is now there.
1150 table = self.getExistingTable(name, spec)
1151 if table is None: 1151 ↛ 1152line 1151 didn't jump to line 1152 because the condition on line 1151 was never true
1152 raise
1153 return table
1155 def getExistingTable(self, name: str, spec: ddl.TableSpec) -> sqlalchemy.schema.Table | None:
1156 """Obtain an existing table with the given name and specification.
1158 Parameters
1159 ----------
1160 name : `str`
1161 Name of the table (not including namespace qualifiers).
1162 spec : `TableSpec`
1163 Specification for the table. This will be used when creating the
1164 SQLAlchemy representation of the table, and it is used to
1165 check that the actual table in the database is consistent.
1167 Returns
1168 -------
1169 table : `sqlalchemy.schema.Table` or `None`
1170 SQLAlchemy representation of the table, or `None` if it does not
1171 exist.
1173 Raises
1174 ------
1175 DatabaseConflictError
1176 Raised if the table exists but ``spec`` is inconsistent with its
1177 definition.
1179 Notes
1180 -----
1181 This method can be called within transactions and never modifies the
1182 database.
1184 Subclasses may override this method, but usually should not need to.
1185 """
1186 assert self._metadata is not None, "Static tables must be declared before dynamic tables."
1187 table = self._metadata.get_table(name)
1188 if table is not None:
1189 if spec.fields.names != set(table.columns.keys()):
1190 raise DatabaseConflictError(
1191 f"Table '{name}' has already been defined differently; the new "
1192 f"specification has columns {list(spec.fields.names)}, while "
1193 f"the previous definition has {list(table.columns.keys())}."
1194 )
1195 else:
1196 inspector = sqlalchemy.inspect(
1197 self._engine if self._session_connection is None else self._session_connection, raiseerr=True
1198 )
1199 if name in inspector.get_table_names(schema=self.namespace):
1200 _checkExistingTableDefinition(name, spec, inspector.get_columns(name, schema=self.namespace))
1201 return self._metadata.add_table(self, name, spec)
1202 return table
1204 def _make_temporary_table(
1205 self,
1206 connection: sqlalchemy.engine.Connection,
1207 spec: ddl.TableSpec,
1208 name: str | None = None,
1209 **kwargs: Any,
1210 ) -> sqlalchemy.schema.Table:
1211 """Create a temporary table.
1213 Parameters
1214 ----------
1215 connection : `sqlalchemy.engine.Connection`
1216 Connection to use when creating the table.
1217 spec : `TableSpec`
1218 Specification for the table.
1219 name : `str`, optional
1220 A unique (within this session/connetion) name for the table.
1221 Subclasses may override to modify the actual name used. If not
1222 provided, a unique name will be generated.
1223 **kwargs
1224 Additional keyword arguments to forward to the
1225 `sqlalchemy.schema.Table` constructor. This is provided to make it
1226 easier for derived classes to delegate to ``super()`` while making
1227 only minor changes.
1229 Returns
1230 -------
1231 table : `sqlalchemy.schema.Table`
1232 SQLAlchemy representation of the table.
1233 """
1234 if name is None:
1235 name = f"tmp_{uuid.uuid4().hex}"
1236 if self._metadata is None: 1236 ↛ 1237line 1236 didn't jump to line 1237 because the condition on line 1236 was never true
1237 raise RuntimeError("Cannot create temporary table before static schema is defined.")
1238 table = self._metadata.add_table(
1239 self, name, spec, prefixes=["TEMPORARY"], schema=sqlalchemy.schema.BLANK_SCHEMA, **kwargs
1240 )
1241 if table.key in self._temp_tables and table.key != name: 1241 ↛ 1242line 1241 didn't jump to line 1242 because the condition on line 1241 was never true
1242 raise ValueError(
1243 f"A temporary table with name {name} (transformed to {table.key} by Database) already exists."
1244 )
1245 with self._transaction():
1246 table.create(connection)
1247 return table
1249 @classmethod
1250 def getTimespanRepresentation(cls) -> type[TimespanDatabaseRepresentation]:
1251 """Return a `type` that encapsulates the way `Timespan` objects are
1252 stored in this database.
1254 `Database` does not automatically use the return type of this method
1255 anywhere else; calling code is responsible for making sure that DDL
1256 and queries are consistent with it.
1258 Returns
1259 -------
1260 TimespanReprClass : `type` (`TimespanDatabaseRepresention` subclass)
1261 A type that encapsulates the way `Timespan` objects should be
1262 stored in this database.
1264 Notes
1265 -----
1266 There are two big reasons we've decided to keep timespan-mangling logic
1267 outside the `Database` implementations, even though the choice of
1268 representation is ultimately up to a `Database` implementation:
1270 - Timespans appear in relatively few tables and queries in our
1271 typical usage, and the code that operates on them is already aware
1272 that it is working with timespans. In contrast, a
1273 timespan-representation-aware implementation of, say,
1274 `Database.insert`, would need to have extra logic to identify when
1275 timespan-mangling needed to occur, which would usually be useless
1276 overhead.
1278 - SQLAlchemy's rich SELECT query expression system has no way to wrap
1279 multiple columns in a single expression object (the ORM does, but
1280 we are not using the ORM). So we would have to wrap _much_ more of
1281 that code in our own interfaces to encapsulate timespan
1282 representations there.
1283 """
1284 return TimespanDatabaseRepresentation.Compound
1286 def sync(
1287 self,
1288 table: sqlalchemy.schema.Table,
1289 *,
1290 keys: dict[str, Any],
1291 compared: dict[str, Any] | None = None,
1292 extra: dict[str, Any] | None = None,
1293 returning: Sequence[str] | None = None,
1294 update: bool = False,
1295 ) -> tuple[dict[str, Any] | None, bool | dict[str, Any]]:
1296 """Insert into a table as necessary to ensure database contains
1297 values equivalent to the given ones.
1299 Parameters
1300 ----------
1301 table : `sqlalchemy.schema.Table`
1302 Table to be queried and possibly inserted into.
1303 keys : `dict`
1304 Column name-value pairs used to search for an existing row; must
1305 be a combination that can be used to select a single row if one
1306 exists. If such a row does not exist, these values are used in
1307 the insert.
1308 compared : `dict`, optional
1309 Column name-value pairs that are compared to those in any existing
1310 row. If such a row does not exist, these rows are used in the
1311 insert.
1312 extra : `dict`, optional
1313 Column name-value pairs that are ignored if a matching row exists,
1314 but used in an insert if one is necessary.
1315 returning : `~collections.abc.Sequence` of `str`, optional
1316 The names of columns whose values should be returned.
1317 update : `bool`, optional
1318 If `True` (`False` is default), update the existing row with the
1319 values in ``compared`` instead of raising `DatabaseConflictError`.
1321 Returns
1322 -------
1323 row : `dict`, optional
1324 The value of the fields indicated by ``returning``, or `None` if
1325 ``returning`` is `None`.
1326 inserted_or_updated : `bool` or `dict`
1327 If `True`, a new row was inserted; if `False`, a matching row
1328 already existed. If a `dict` (only possible if ``update=True``),
1329 then an existing row was updated, and the dict maps the names of
1330 the updated columns to their *old* values (new values can be
1331 obtained from ``compared``).
1333 Raises
1334 ------
1335 DatabaseConflictError
1336 Raised if the values in ``compared`` do not match the values in the
1337 database.
1338 ReadOnlyDatabaseError
1339 Raised if `isWriteable` returns `False`, and no matching record
1340 already exists.
1342 Notes
1343 -----
1344 May be used inside transaction contexts, so implementations may not
1345 perform operations that interrupt transactions.
1347 It may be called on read-only databases if and only if the matching row
1348 does in fact already exist.
1349 """
1351 def check() -> tuple[int, dict[str, Any] | None, list | None]:
1352 """Query for a row that matches the ``key`` argument, and compare
1353 to what was given by the caller.
1355 Returns
1356 -------
1357 n : `int`
1358 Number of matching rows. ``n != 1`` is always an error, but
1359 it's a different kind of error depending on where `check` is
1360 being called.
1361 bad : `dict` or `None`
1362 The subset of the keys of ``compared`` for which the existing
1363 values did not match the given one, mapped to the existing
1364 values in the database. Once again, ``not bad`` is always an
1365 error, but a different kind on context. `None` if ``n != 1``.
1366 result : `list` or `None`
1367 Results in the database that correspond to the columns given
1368 in ``returning``, or `None` if ``returning is None``.
1369 """
1370 toSelect: set[str] = set()
1371 if compared is not None:
1372 toSelect.update(compared.keys())
1373 if returning is not None:
1374 toSelect.update(returning)
1375 if not toSelect:
1376 # Need to select some column, even if we just want to see
1377 # how many rows we get back.
1378 toSelect.add(next(iter(keys.keys())))
1379 selectSql = (
1380 sqlalchemy.sql.select(*[table.columns[k].label(k) for k in toSelect])
1381 .select_from(table)
1382 .where(sqlalchemy.sql.and_(*[table.columns[k] == v for k, v in keys.items()]))
1383 )
1384 with self._transaction() as (_, connection):
1385 fetched = list(connection.execute(selectSql).mappings())
1386 if len(fetched) != 1:
1387 return len(fetched), None, None
1388 existing = fetched[0]
1389 if compared is not None:
1391 def safeNotEqual(a: Any, b: Any) -> bool:
1392 if isinstance(a, astropy.time.Time): 1392 ↛ 1393line 1392 didn't jump to line 1393 because the condition on line 1392 was never true
1393 return not time_utils.TimeConverter().times_equal(a, b)
1394 return a != b
1396 inconsistencies = {
1397 k: existing[k] for k, v in compared.items() if safeNotEqual(existing[k], v)
1398 }
1399 else:
1400 inconsistencies = {}
1401 if returning is not None:
1402 toReturn: list | None = [existing[k] for k in returning]
1403 else:
1404 toReturn = None
1405 return 1, inconsistencies, toReturn
1407 def _format_bad(inconsistencies: dict[str, Any]) -> str:
1408 """Format the 'bad' dictionary of existing values returned by
1409 ``check`` into a string suitable for an error message.
1410 """
1411 assert compared is not None, "Should not be able to get inconsistencies without comparing."
1412 return ", ".join(f"{k}: {v!r} != {compared[k]!r}" for k, v in inconsistencies.items())
1414 if self.isTableWriteable(table):
1415 # Try an insert first, but allow it to fail (in only specific
1416 # ways).
1417 row = keys.copy()
1418 if compared is not None:
1419 row.update(compared)
1420 if extra is not None:
1421 row.update(extra)
1422 with self.transaction():
1423 inserted = bool(self.ensure(table, row))
1424 inserted_or_updated: bool | dict[str, Any]
1425 # Need to perform check() for this branch inside the
1426 # transaction, so we roll back an insert that didn't do
1427 # what we expected. That limits the extent to which we
1428 # can reduce duplication between this block and the other
1429 # ones that perform similar logic.
1430 n, bad, result = check()
1431 if n < 1: 1431 ↛ 1432line 1431 didn't jump to line 1432 because the condition on line 1431 was never true
1432 raise ConflictingDefinitionError(
1433 f"Attempted to ensure {row} exists by inserting it with ON CONFLICT IGNORE, "
1434 f"but a post-insert query on {keys} returned no results. "
1435 f"Insert was {'' if inserted else 'not '}reported as successful. "
1436 "This can occur if the insert violated a database constraint other than the "
1437 "unique constraint or primary key used to identify the row in this call."
1438 )
1439 elif n > 1: 1439 ↛ 1440line 1439 didn't jump to line 1440 because the condition on line 1439 was never true
1440 raise RuntimeError(
1441 f"Keys passed to sync {keys.keys()} do not comprise a "
1442 f"unique constraint for table {table.name}."
1443 )
1444 elif bad:
1445 assert compared is not None, (
1446 "Should not be able to get inconsistencies without comparing."
1447 )
1448 if inserted: 1448 ↛ 1449line 1448 didn't jump to line 1449 because the condition on line 1448 was never true
1449 raise RuntimeError(
1450 f"Conflict ({bad}) in sync after successful insert; this is "
1451 "possible if the same table is being updated by a concurrent "
1452 "process that isn't using sync, but it may also be a bug in "
1453 "daf_butler."
1454 )
1455 elif update:
1456 with self._transaction() as (_, connection):
1457 connection.execute(
1458 table.update()
1459 .where(sqlalchemy.sql.and_(*[table.columns[k] == v for k, v in keys.items()]))
1460 .values(**{k: compared[k] for k in bad})
1461 )
1462 inserted_or_updated = bad
1463 else:
1464 raise DatabaseConflictError(
1465 f"Conflict in sync for table {table.name} on column(s) {_format_bad(bad)}."
1466 )
1467 else:
1468 inserted_or_updated = inserted
1469 else:
1470 # Database is not writeable; just see if the row exists.
1471 n, bad, result = check()
1472 if n < 1:
1473 raise ReadOnlyDatabaseError("sync needs to insert, but database is read-only.")
1474 elif n > 1: 1474 ↛ 1475line 1474 didn't jump to line 1475 because the condition on line 1474 was never true
1475 raise RuntimeError("Keys passed to sync do not comprise a unique constraint.")
1476 elif bad: 1476 ↛ 1477line 1476 didn't jump to line 1477 because the condition on line 1476 was never true
1477 if update:
1478 raise ReadOnlyDatabaseError("sync needs to update, but database is read-only.")
1479 else:
1480 raise DatabaseConflictError(
1481 f"Conflict in sync for table {table.name} on column(s) {_format_bad(bad)}."
1482 )
1483 inserted_or_updated = False
1484 if returning is None:
1485 return None, inserted_or_updated
1486 else:
1487 assert result is not None
1488 return dict(zip(returning, result, strict=True)), inserted_or_updated
1490 def insert(
1491 self,
1492 table: sqlalchemy.schema.Table,
1493 *rows: dict,
1494 returnIds: bool = False,
1495 select: sqlalchemy.sql.expression.SelectBase | None = None,
1496 names: Iterable[str] | None = None,
1497 ) -> list[int] | None:
1498 """Insert one or more rows into a table, optionally returning
1499 autoincrement primary key values.
1501 Parameters
1502 ----------
1503 table : `sqlalchemy.schema.Table`
1504 Table rows should be inserted into.
1505 *rows : `dict`
1506 Positional arguments are the rows to be inserted, as dictionaries
1507 mapping column name to value. The keys in all dictionaries must
1508 be the same.
1509 returnIds : `bool`, optional
1510 If `True` (`False` is default), return the values of the table's
1511 autoincrement primary key field (which much exist).
1512 select : `sqlalchemy.sql.SelectBase`, optional
1513 A SELECT query expression to insert rows from. Cannot be provided
1514 with either ``rows`` or ``returnIds=True``.
1515 names : `~collections.abc.Iterable` [ `str` ], optional
1516 Names of columns in ``table`` to be populated, ordered to match the
1517 columns returned by ``select``. Ignored if ``select`` is `None`.
1518 If not provided, the columns returned by ``select`` must be named
1519 to match the desired columns of ``table``.
1521 Returns
1522 -------
1523 ids : `None`, or `list` of `int`
1524 If ``returnIds`` is `True`, a `list` containing the inserted
1525 values for the table's autoincrement primary key.
1527 Raises
1528 ------
1529 ReadOnlyDatabaseError
1530 Raised if `isWriteable` returns `False` when this method is called.
1532 Notes
1533 -----
1534 The default implementation uses bulk insert syntax when ``returnIds``
1535 is `False`, and a loop over single-row insert operations when it is
1536 `True`.
1538 Derived classes should reimplement when they can provide a more
1539 efficient implementation (especially for the latter case).
1541 May be used inside transaction contexts, so implementations may not
1542 perform operations that interrupt transactions.
1543 """
1544 self.assertTableWriteable(table, f"Cannot insert into read-only table {table}.")
1545 if select is not None and (rows or returnIds): 1545 ↛ 1546line 1545 didn't jump to line 1546 because the condition on line 1545 was never true
1546 raise TypeError("'select' is incompatible with passing value rows or returnIds=True.")
1547 if not rows and select is None:
1548 if returnIds: 1548 ↛ 1549line 1548 didn't jump to line 1549 because the condition on line 1548 was never true
1549 return []
1550 else:
1551 return None
1552 with self._transaction() as (_, connection):
1553 if not returnIds:
1554 if select is not None:
1555 if names is None:
1556 names = select.selected_columns.keys()
1557 connection.execute(table.insert().from_select(list(names), select))
1558 else:
1559 connection.execute(table.insert(), rows)
1560 return None
1561 else:
1562 sql = table.insert()
1563 ids = []
1564 for row in rows:
1565 key = connection.execute(sql, row).inserted_primary_key
1566 assert key is not None
1567 ids.append(key[0])
1568 return ids
1570 @abstractmethod
1571 def replace(self, table: sqlalchemy.schema.Table, *rows: dict) -> None:
1572 """Insert one or more rows into a table, replacing any existing rows
1573 for which insertion of a new row would violate the primary key
1574 constraint.
1576 Parameters
1577 ----------
1578 table : `sqlalchemy.schema.Table`
1579 Table rows should be inserted into.
1580 *rows
1581 Positional arguments are the rows to be inserted, as dictionaries
1582 mapping column name to value. The keys in all dictionaries must
1583 be the same.
1585 Raises
1586 ------
1587 ReadOnlyDatabaseError
1588 Raised if `isWriteable` returns `False` when this method is called.
1590 Notes
1591 -----
1592 May be used inside transaction contexts, so implementations may not
1593 perform operations that interrupt transactions.
1595 Implementations should raise a `sqlalchemy.exc.IntegrityError`
1596 exception when a constraint other than the primary key would be
1597 violated.
1599 Implementations are not required to support `replace` on tables
1600 with autoincrement keys.
1601 """
1602 raise NotImplementedError()
1604 @abstractmethod
1605 def ensure(self, table: sqlalchemy.schema.Table, *rows: dict, primary_key_only: bool = False) -> int:
1606 """Insert one or more rows into a table, skipping any rows for which
1607 insertion would violate a unique constraint.
1609 Parameters
1610 ----------
1611 table : `sqlalchemy.schema.Table`
1612 Table rows should be inserted into.
1613 *rows
1614 Positional arguments are the rows to be inserted, as dictionaries
1615 mapping column name to value. The keys in all dictionaries must
1616 be the same.
1617 primary_key_only : `bool`, optional
1618 If `True` (`False` is default), only skip rows that violate the
1619 primary key constraint, and raise an exception (and rollback
1620 transactions) for other constraint violations.
1622 Returns
1623 -------
1624 count : `int`
1625 The number of rows actually inserted.
1627 Raises
1628 ------
1629 ReadOnlyDatabaseError
1630 Raised if `isWriteable` returns `False` when this method is called.
1631 This is raised even if the operation would do nothing even on a
1632 writeable database.
1634 Notes
1635 -----
1636 May be used inside transaction contexts, so implementations may not
1637 perform operations that interrupt transactions.
1639 Implementations are not required to support `ensure` on tables
1640 with autoincrement keys.
1641 """
1642 raise NotImplementedError()
1644 def delete(self, table: sqlalchemy.schema.Table, columns: Iterable[str], *rows: dict) -> int:
1645 """Delete one or more rows from a table.
1647 Parameters
1648 ----------
1649 table : `sqlalchemy.schema.Table`
1650 Table that rows should be deleted from.
1651 columns : `~collections.abc.Iterable` of `str`
1652 The names of columns that will be used to constrain the rows to
1653 be deleted; these will be combined via ``AND`` to form the
1654 ``WHERE`` clause of the delete query.
1655 *rows
1656 Positional arguments are the keys of rows to be deleted, as
1657 dictionaries mapping column name to value. The keys in all
1658 dictionaries must be exactly the names in ``columns``.
1660 Returns
1661 -------
1662 count : `int`
1663 Number of rows deleted.
1665 Raises
1666 ------
1667 ReadOnlyDatabaseError
1668 Raised if `isWriteable` returns `False` when this method is called.
1670 Notes
1671 -----
1672 May be used inside transaction contexts, so implementations may not
1673 perform operations that interrupt transactions.
1675 The default implementation should be sufficient for most derived
1676 classes.
1677 """
1678 self.assertTableWriteable(table, f"Cannot delete from read-only table {table}.")
1679 if columns and not rows: 1679 ↛ 1685line 1679 didn't jump to line 1685 because the condition on line 1679 was never true
1680 # If there are no columns, this operation is supposed to delete
1681 # everything (so we proceed as usual). But if there are columns,
1682 # but no rows, it was a constrained bulk operation where the
1683 # constraint is that no rows match, and we should short-circuit
1684 # while reporting that no rows were affected.
1685 return 0
1686 sql = table.delete()
1687 columns = list(columns) # Force iterators to list
1689 # More efficient to use IN operator if there is only one
1690 # variable changing across all rows.
1691 content: dict[str, set] = defaultdict(set)
1692 if len(columns) == 1:
1693 # Nothing to calculate since we can always use IN
1694 column = columns[0]
1695 changing_columns = [column]
1696 content[column] = {row[column] for row in rows}
1697 else:
1698 for row in rows:
1699 for k, v in row.items():
1700 content[k].add(v)
1701 changing_columns = [col for col, values in content.items() if len(values) > 1]
1703 if not columns:
1704 # No columns means to delete all rows.
1705 with self._transaction() as (_, connection):
1706 return connection.execute(sql).rowcount
1707 elif len(changing_columns) != 1:
1708 # More than one column changes each time so do explicit bind
1709 # parameters and have each row processed separately.
1710 whereTerms = [table.columns[name] == sqlalchemy.sql.bindparam(name) for name in columns]
1711 if whereTerms: 1711 ↛ 1713line 1711 didn't jump to line 1713 because the condition on line 1711 was always true
1712 sql = sql.where(sqlalchemy.sql.and_(*whereTerms))
1713 with self._transaction() as (_, connection):
1714 return connection.execute(sql, rows).rowcount
1715 else:
1716 # One of the columns has changing values but any others are
1717 # fixed. In this case we can use an IN operator and be more
1718 # efficient.
1719 name = changing_columns.pop()
1721 # Simple where clause for the unchanging columns
1722 clauses = []
1723 for k, v in content.items():
1724 if k == name:
1725 continue
1726 # The set only has one element
1727 clauses.append(table.columns[k] == v.pop())
1729 # The IN operator will not work for "infinite" numbers of
1730 # rows so must batch it up into distinct calls.
1731 in_content = list(content[name])
1732 n_elements = len(in_content)
1734 rowcount = 0
1735 iposn = 0
1736 n_per_loop = 1_000 # Controls how many items to put in IN clause
1737 with self._transaction() as (_, connection):
1738 for iposn in range(0, n_elements, n_per_loop):
1739 endpos = iposn + n_per_loop
1740 in_clause = self.make_in_array_constraint(table.columns[name], in_content[iposn:endpos])
1742 newsql = sql.where(sqlalchemy.sql.and_(*clauses, in_clause))
1743 rowcount += connection.execute(newsql).rowcount
1744 return rowcount
1746 def deleteWhere(self, table: sqlalchemy.schema.Table, where: sqlalchemy.sql.ColumnElement) -> int:
1747 """Delete rows from a table with pre-constructed WHERE clause.
1749 Parameters
1750 ----------
1751 table : `sqlalchemy.schema.Table`
1752 Table that rows should be deleted from.
1753 where : `sqlalchemy.sql.ClauseElement`
1754 The names of columns that will be used to constrain the rows to
1755 be deleted; these will be combined via ``AND`` to form the
1756 ``WHERE`` clause of the delete query.
1758 Returns
1759 -------
1760 count : `int`
1761 Number of rows deleted.
1763 Raises
1764 ------
1765 ReadOnlyDatabaseError
1766 Raised if `isWriteable` returns `False` when this method is called.
1768 Notes
1769 -----
1770 May be used inside transaction contexts, so implementations may not
1771 perform operations that interrupt transactions.
1773 The default implementation should be sufficient for most derived
1774 classes.
1775 """
1776 self.assertTableWriteable(table, f"Cannot delete from read-only table {table}.")
1778 sql = table.delete().where(where)
1779 with self._transaction() as (_, connection):
1780 return connection.execute(sql).rowcount
1782 def update(self, table: sqlalchemy.schema.Table, where: dict[str, str], *rows: dict) -> int:
1783 """Update one or more rows in a table.
1785 Parameters
1786 ----------
1787 table : `sqlalchemy.schema.Table`
1788 Table containing the rows to be updated.
1789 where : `dict` [`str`, `str`]
1790 A mapping from the names of columns that will be used to search for
1791 existing rows to the keys that will hold these values in the
1792 ``rows`` dictionaries. Note that these may not be the same due to
1793 SQLAlchemy limitations.
1794 *rows
1795 Positional arguments are the rows to be updated. The keys in all
1796 dictionaries must be the same, and may correspond to either a
1797 value in the ``where`` dictionary or the name of a column to be
1798 updated.
1800 Returns
1801 -------
1802 count : `int`
1803 Number of rows matched (regardless of whether the update actually
1804 modified them).
1806 Raises
1807 ------
1808 ReadOnlyDatabaseError
1809 Raised if `isWriteable` returns `False` when this method is called.
1811 Notes
1812 -----
1813 May be used inside transaction contexts, so implementations may not
1814 perform operations that interrupt transactions.
1816 The default implementation should be sufficient for most derived
1817 classes.
1818 """
1819 self.assertTableWriteable(table, f"Cannot update read-only table {table}.")
1820 if not rows: 1820 ↛ 1821line 1820 didn't jump to line 1821 because the condition on line 1820 was never true
1821 return 0
1822 sql = table.update().where(
1823 sqlalchemy.sql.and_(*[table.columns[k] == sqlalchemy.sql.bindparam(v) for k, v in where.items()])
1824 )
1825 with self._transaction() as (_, connection):
1826 return connection.execute(sql, rows).rowcount
1828 @contextmanager
1829 def query(
1830 self,
1831 sql: sqlalchemy.sql.expression.Executable | sqlalchemy.sql.expression.SelectBase,
1832 *args: Any,
1833 **kwargs: Any,
1834 ) -> Iterator[sqlalchemy.engine.CursorResult]:
1835 """Run a SELECT query against the database.
1837 Parameters
1838 ----------
1839 sql : `sqlalchemy.sql.expression.SelectBase`
1840 A SQLAlchemy representation of a ``SELECT`` query.
1841 *args
1842 Additional positional arguments are forwarded to
1843 `sqlalchemy.engine.Connection.execute`.
1844 **kwargs
1845 Additional keyword arguments are forwarded to
1846 `sqlalchemy.engine.Connection.execute`.
1848 Returns
1849 -------
1850 result_context : `sqlalchemy.engine.CursorResults`
1851 Context manager that returns the query result object when entered.
1852 These results are invalidated when the context is exited.
1853 """
1854 if self._session_connection is None:
1855 connection = self._engine.connect()
1856 else:
1857 connection = self._session_connection
1858 try:
1859 self._log_query_and_plan(connection, sql)
1860 # TODO: SelectBase is not good for execute(), but it used
1861 # everywhere, e.g. in daf_relation. We should switch to Executable
1862 # at some point.
1863 with connection.execute(
1864 cast(sqlalchemy.sql.expression.Executable, sql), *args, **kwargs
1865 ) as result:
1866 yield result
1867 finally:
1868 if connection is not self._session_connection:
1869 connection.close()
1871 def _log_query_and_plan(
1872 self,
1873 connection: sqlalchemy.Connection,
1874 sql: sqlalchemy.sql.expression.Executable | sqlalchemy.sql.expression.SelectBase,
1875 ) -> None:
1876 """Log the given SQL statement and the DB's plan for executing it if
1877 the environment variable DAF_BUTLER_DEBUG_QUERIES is set to a truthy
1878 value.
1879 """
1880 if os.environ.get("DAF_BUTLER_DEBUG_QUERIES", False): 1880 ↛ 1881line 1880 didn't jump to line 1881 because the condition on line 1880 was never true
1881 assert isinstance(sql, sqlalchemy.SelectBase)
1882 compiled = sql.compile(connection)
1883 print(
1884 f"Executing SQL statement:\n{compiled}\nBind parameters: {compiled.params}", file=sys.stderr
1885 )
1887 query_plan = get_query_plan(connection, sql)
1888 print(f"Query plan:\n{query_plan}", file=sys.stderr)
1890 @abstractmethod
1891 def constant_rows(
1892 self,
1893 fields: NamedValueAbstractSet[ddl.FieldSpec],
1894 *rows: dict,
1895 name: str | None = None,
1896 ) -> sqlalchemy.sql.FromClause:
1897 """Return a SQLAlchemy object that represents a small number of
1898 constant-valued rows.
1900 Parameters
1901 ----------
1902 fields : `NamedValueAbstractSet` [ `ddl.FieldSpec` ]
1903 The columns of the rows. Unique and foreign key constraints are
1904 ignored.
1905 *rows : `dict`
1906 Values for the rows.
1907 name : `str`, optional
1908 If provided, the name of the SQL construct. If not provided, an
1909 opaque but unique identifier is generated.
1911 Returns
1912 -------
1913 from_clause : `sqlalchemy.sql.FromClause`
1914 SQLAlchemy object representing the given rows. This is guaranteed
1915 to be something that can be directly joined into a ``SELECT``
1916 query's ``FROM`` clause, and will not involve a temporary table
1917 that needs to be cleaned up later.
1919 Notes
1920 -----
1921 The default implementation uses the SQL-standard ``VALUES`` construct,
1922 but support for that construct is varied enough across popular RDBMSs
1923 that the method is still marked abstract to force explicit opt-in via
1924 delegation to `super`.
1925 """
1926 if name is None:
1927 name = f"tmp_{uuid.uuid4().hex}"
1928 return sqlalchemy.sql.values(
1929 *[sqlalchemy.Column(field.name, field.getSizedColumnType()) for field in fields],
1930 name=name,
1931 ).data([tuple(row[name] for name in fields.names) for row in rows])
1933 def get_constant_rows_max(self) -> int:
1934 """Return the maximum number of rows that should be passed to
1935 `constant_rows` for this backend.
1937 Returns
1938 -------
1939 max : `int`
1940 Maximum number of rows.
1942 Notes
1943 -----
1944 This should reflect typical performance profiles (or a guess at these),
1945 not just hard database engine limits.
1946 """
1947 return 1000
1949 @abstractmethod
1950 def glob_expression(
1951 self, expression: sqlalchemy.ColumnElement[Any], pattern: str
1952 ) -> sqlalchemy.ColumnElement[bool]:
1953 """Produce boolean expression for expression match against glob-like
1954 pattern.
1956 Parameters
1957 ----------
1958 expression : `sqlalchemy.ColumnElement`
1959 Column expression that evaluates to string.
1960 pattern : `str`
1961 GLob pattern string.
1963 Returns
1964 -------
1965 glob : `sqlalchemy.ColumnElement`
1966 Boolean expression that matches ``expression`` against ``pattern``.
1967 """
1968 raise NotImplementedError()
1970 @property
1971 @abstractmethod
1972 def has_distinct_on(self) -> bool:
1973 """Whether this database supports the ``DISTINCT ON`` SQL construct."""
1974 raise NotImplementedError()
1976 @property
1977 @abstractmethod
1978 def has_any_aggregate(self) -> bool:
1979 """Whether this database supports the ``ANY_VALUE`` aggregate function
1980 or something equivalent.
1981 """
1982 raise NotImplementedError()
1984 @property
1985 def supports_temporary_tables(self) -> bool:
1986 return self._allow_temporary_tables
1988 @abstractmethod
1989 def apply_any_aggregate(self, column: sqlalchemy.ColumnElement[Any]) -> sqlalchemy.ColumnElement[Any]:
1990 """Wrap the given SQLAlchemy column in the ``ANY_VALUE`` aggregate
1991 function or its equivalent.
1993 Parameters
1994 ----------
1995 column : `sqlalchemy.ColumnElement`
1996 Original column to wrap.
1998 Returns
1999 -------
2000 wrapped : `sqlalchemy.ColumnElement`
2001 A column element of the same SQL type that can appear in the
2002 ``SELECT`` clause even when this column does not appear in the
2003 ``GROUP BY`` clause.
2005 Notes
2006 -----
2007 This method's behavior is unspecified when `has_any_aggregate` is
2008 `False`; the caller is responsible for checking that property first.
2009 """
2010 raise NotImplementedError()
2012 def make_in_array_constraint(
2013 self, column: sqlalchemy.ColumnElement[_T], values: Iterable[_T]
2014 ) -> sqlalchemy.ColumnElement[bool]:
2015 return column.in_(values)
2017 origin: int
2018 """An integer ID that should be used as the default for any datasets,
2019 quanta, or other entities that use a (autoincrement, origin) compound
2020 primary key (`int`).
2021 """
2023 namespace: str | None
2024 """The schema or namespace this database instance is associated with
2025 (`str` or `None`).
2026 """
2028 name_shrinker: NameShrinker
2029 """An object that can be used to shrink field names to fit within the
2030 identifier limit of the database engine (`NameShrinker`).
2031 """
2034class DatabaseMetadata:
2035 """Wrapper around SqlAlchemy MetaData object to ensure threadsafety.
2037 Parameters
2038 ----------
2039 namespace : `str` or `None`
2040 Name of the schema or namespace this instance is associated with.
2042 Notes
2043 -----
2044 `sqlalchemy.MetaData` is documented to be threadsafe for reads, but not
2045 with concurrent modifications. We add tables dynamically at runtime,
2046 and the MetaData object is shared by all Database instances sharing
2047 the same connection pool.
2048 """
2050 def __init__(self, namespace: str | None) -> None:
2051 self._lock = Lock()
2052 self._metadata = sqlalchemy.MetaData(schema=namespace)
2053 self._tables: dict[str, sqlalchemy.Table] = {}
2055 def add_table(
2056 self, db: Database, name: str, spec: ddl.TableSpec, **kwargs: Any
2057 ) -> sqlalchemy.schema.Table:
2058 """Add a new table to the MetaData object, returning its sqlalchemy
2059 representation. This does not physically create the table in the
2060 database -- it only sets up its definition.
2062 Parameters
2063 ----------
2064 db : `Database`
2065 Database connection associated with the table definition.
2066 name : `str`
2067 The name of the table.
2068 spec : `ddl.TableSpec`
2069 The specification of the table.
2070 **kwargs
2071 Additional keyword arguments to forward to the
2072 `sqlalchemy.schema.Table` constructor.
2074 Returns
2075 -------
2076 table : `sqlalchemy.schema.Table`
2077 The created table.
2078 """
2079 with self._lock:
2080 if (table := self._tables.get(name)) is not None: 2080 ↛ 2081line 2080 didn't jump to line 2081 because the condition on line 2080 was never true
2081 return table
2083 table = db._convertTableSpec(name, spec, self._metadata, **kwargs)
2084 for foreignKeySpec in spec.foreignKeys:
2085 table.append_constraint(db._convertForeignKeySpec(name, foreignKeySpec, self._metadata))
2087 self._tables[name] = table
2088 return table
2090 def get_table(self, name: str) -> sqlalchemy.schema.Table | None:
2091 """Return the definition of a table that was previously added to this
2092 MetaData object.
2094 Parameters
2095 ----------
2096 name : `str`
2097 Name of the table.
2099 Returns
2100 -------
2101 table : `sqlalchemy.schema.Table` or `None`
2102 The table definition, or `None` if the table is not known to this
2103 MetaData instance.
2104 """
2105 with self._lock:
2106 return self._tables.get(name)
2108 def remove_table(self, name: str) -> None:
2109 """Remove a table that was previously added to this MetaData object.
2111 Parameters
2112 ----------
2113 name : `str`
2114 Name of the table.
2115 """
2116 with self._lock:
2117 table = self._tables.pop(name, None)
2118 if table is not None: 2118 ↛ exitline 2118 didn't jump to the function exit
2119 self._metadata.remove(table)
2121 def create_all(self, connection: sqlalchemy.engine.Connection) -> None:
2122 """Create all tables known to this MetaData object in the database.
2123 Same as `sqlalchemy.MetaData.create_all`.
2125 Parameters
2126 ----------
2127 connection : `sqlalchemy.engine.connection`
2128 Database connection that will be used to create tables.
2129 """
2130 with self._lock:
2131 self._metadata.create_all(connection)