|
3 | 3 | from __future__ import annotations |
4 | 4 |
|
5 | 5 | import re |
6 | | -from collections.abc import Iterator |
7 | | -from contextlib import contextmanager |
8 | | -from contextvars import ContextVar, Token |
9 | 6 | from pathlib import Path |
10 | 7 |
|
11 | 8 | from sqlbuild.adapter.discovery.main.session_warehouse_support import ( |
12 | 9 | adapter_supports_session_warehouse, |
13 | 10 | ) |
| 11 | +from sqlbuild.cli.commands._helpers.runtime.warehouse_scope import ( |
| 12 | + current_command_warehouse_scope, |
| 13 | +) |
14 | 14 | from sqlbuild.cli.commands.constants import ( |
15 | 15 | CLI_WAREHOUSE_SOURCE, |
16 | | - COMMAND_WAREHOUSE_GROUPS, |
17 | 16 | CONNECTION_DEFAULT_WAREHOUSE_SOURCE, |
18 | 17 | CONNECTION_WAREHOUSE_SOURCE, |
19 | 18 | ) |
20 | 19 | from sqlbuild.cli.commands.exceptions import CliUserError |
21 | | -from sqlbuild.cli.commands.types import CliCommand |
22 | 20 | from sqlbuild.cli.entry.models import ( |
23 | 21 | AuthoredTargetWarehouse, |
24 | 22 | CommandWarehouseScope, |
|
47 | 45 | _CONNECTION_SOURCES: frozenset[str] = frozenset( |
48 | 46 | {CONNECTION_WAREHOUSE_SOURCE, CONNECTION_DEFAULT_WAREHOUSE_SOURCE} |
49 | 47 | ) |
50 | | -_CURRENT_SCOPE: ContextVar[CommandWarehouseScope | None] = ContextVar( |
51 | | - "sqlbuild_command_warehouse_scope", default=None |
52 | | -) |
53 | | - |
54 | | - |
55 | | -@contextmanager |
56 | | -def command_warehouse_scope( |
57 | | - *, |
58 | | - command: CliCommand | str | None, |
59 | | - cli_warehouse: str | None, |
60 | | - cli_vars: dict[str, object] | None = None, |
61 | | -) -> Iterator[CommandWarehouseScope]: |
62 | | - """Install the warehouse group, CLI override, and CLI vars for one command invocation.""" |
63 | | - |
64 | | - scope: CommandWarehouseScope = CommandWarehouseScope( |
65 | | - group=None if command is None else COMMAND_WAREHOUSE_GROUPS.get(CliCommand(command)), |
66 | | - cli_warehouse=cli_warehouse, |
67 | | - cli_vars={} if cli_vars is None else dict(cli_vars), |
68 | | - ) |
69 | | - token: Token[CommandWarehouseScope | None] = _CURRENT_SCOPE.set(scope) |
70 | | - try: |
71 | | - yield scope |
72 | | - finally: |
73 | | - _CURRENT_SCOPE.reset(token) |
74 | | - |
75 | | - |
76 | | -def current_command_warehouse_scope() -> CommandWarehouseScope: |
77 | | - """Return the active command's warehouse scope, or an empty one outside the CLI.""" |
78 | | - |
79 | | - scope: CommandWarehouseScope | None = _CURRENT_SCOPE.get() |
80 | | - return scope if scope is not None else CommandWarehouseScope(group=None) |
81 | 48 |
|
82 | 49 |
|
83 | 50 | def validate_cli_warehouse(*, cli_warehouse: str | None, project_dir: Path) -> None: |
|
0 commit comments