dbt-sqlserver/dbt/adapters/sqlserver/sql_server_adapter.py at 4b5829803f00ac11a35b859147acabbdd3714d35 · ms32035/dbt-sqlserver · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
# https://github.com/microsoft/dbt-fabric/blob/main/dbt/adapters/fabric/fabric_adapter.py
from typing import Optional

import dbt.exceptions
from dbt.adapters.base import ConstraintSupport
from dbt.adapters.fabric import FabricAdapter
from dbt.contracts.graph.nodes import ConstraintType, ModelLevelConstraint

from dbt.adapters.sqlserver.sql_server_column import SQLServerColumn
from dbt.adapters.sqlserver.sql_server_configs import SQLServerConfigs
from dbt.adapters.sqlserver.sql_server_connection_manager import SQLServerConnectionManager

# from dbt.adapters.capability import Capability, CapabilityDict, CapabilitySupport, Support


class SQLServerAdapter(FabricAdapter):
    ConnectionManager = SQLServerConnectionManager
    Column = SQLServerColumn
    AdapterSpecificConfigs = SQLServerConfigs

    # _capabilities: CapabilityDict = CapabilityDict(
    #     {
    #         Capability.SchemaMetadataByRelations: CapabilitySupport(support=Support.Full),
    #         Capability.TableLastModifiedMetadata: CapabilitySupport(support=Support.Full),
    #     }
    # )

    # region - these are implement in fabric but not in sqlserver
    # _capabilities: CapabilityDict = CapabilityDict(
    #     {
    #         Capability.SchemaMetadataByRelations: CapabilitySupport(support=Support.Full),
    #         Capability.TableLastModifiedMetadata: CapabilitySupport(support=Support.Full),
    #     }
    # )
    CONSTRAINT_SUPPORT = {
        ConstraintType.check: ConstraintSupport.ENFORCED,
        ConstraintType.not_null: ConstraintSupport.ENFORCED,
        ConstraintType.unique: ConstraintSupport.ENFORCED,
        ConstraintType.primary_key: ConstraintSupport.ENFORCED,
        ConstraintType.foreign_key: ConstraintSupport.ENFORCED,
        ConstraintType.custom: ConstraintSupport.ENFORCED,
    }

    # @available.parse(lambda *a, **k: [])
    # def get_column_schema_from_query(self, sql: str) -> List[BaseColumn]:
    #     """Get a list of the Columns with names and data types from the given sql."""
    #     _, cursor = self.connections.add_select_query(sql)

    #     columns = [
    #         self.Column.create(
    #             column_name, self.connections.data_type_code_to_name(column_type_code)
    #         )
    #         # https://peps.python.org/pep-0249/#description
    #         for column_name, column_type_code, *_ in cursor.description
    #     ]
    #     return columns
    # endregion

    @classmethod
    def render_model_constraint(cls, constraint: ModelLevelConstraint) -> Optional[str]:
        constraint_prefix = "add constraint "
        column_list = ", ".join(constraint.columns)

        if constraint.name is None:
            raise dbt.exceptions.DbtDatabaseError(
                "Constraint name cannot be empty. Provide constraint name  - column "
                + column_list
                + " and run the project again."
            )

        if constraint.type == ConstraintType.unique:
            return constraint_prefix + f"{constraint.name} unique nonclustered({column_list})"
        elif constraint.type == ConstraintType.primary_key:
            return constraint_prefix + f"{constraint.name} primary key nonclustered({column_list})"
        elif constraint.type == ConstraintType.foreign_key and constraint.expression:
            return (
                constraint_prefix
                + f"{constraint.name} foreign key({column_list}) references "
                + constraint.expression
            )
        elif constraint.type == ConstraintType.check and constraint.expression:
            return f"{constraint_prefix} {constraint.name} check ({constraint.expression})"
        elif constraint.type == ConstraintType.custom and constraint.expression:
            return f"{constraint_prefix} {constraint.name} {constraint.expression}"
        else:
            return None

    @classmethod
    def date_function(cls):
        return "getdate()"