We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Row count not working for snowflake
**Describe the bug**
Currently row count fails for snowflake 

**Expected behavior**
Row count should work snowflake



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python-sdk/src/astro/table.py`
Content:
```
1 from __future__ import annotations
2 
3 import random
4 import string
5 from typing import Any
6 
7 from attr import define, field, fields_dict
8 from sqlalchemy import Column, MetaData, func, select
9 
10 from astro.airflow.datasets import Dataset
11 from astro.databases import create_database
12 
13 MAX_TABLE_NAME_LENGTH = 62
14 TEMP_PREFIX = "_tmp_"
15 
16 
17 @define
18 class Metadata:
19     """
20     Contains additional information to access a SQL Table, which is very likely optional and, in some cases, may
21     be database-specific.
22 
23     :param schema: A schema name
24     :param database: A database name
25     """
26 
27     # This property is used by several databases, including: Postgres, Snowflake and BigQuery ("namespace")
28     schema: str | None = None
29     database: str | None = None
30 
31     def is_empty(self) -> bool:
32         """Check if all the fields are None."""
33         return all(getattr(self, field_name) is None for field_name in fields_dict(self.__class__))
34 
35 
36 @define(slots=False)
37 class BaseTable:
38     """
39     Base class that has information necessary to access a SQL Table. It is agnostic to the database type.
40     If no name is given, it auto-generates a name for the Table and considers it temporary.
41 
42     Temporary tables are prefixed with the prefix TEMP_PREFIX.
43 
44     :param name: The name of the database table. If name not provided then it would create a temporary name
45     :param conn_id: The Airflow connection id. This will be used to identify the right database type at the runtime
46     :param metadata: A metadata object which will have database or schema name
47     :param columns: columns which define the database table schema.
48     :sphinx-autoapi-skip:
49     """
50 
51     template_fields = ("name",)
52 
53     # TODO: discuss alternative names to this class, since it contains metadata as opposed to be the
54     # SQL table itself
55     # Some ideas: TableRef, TableMetadata, TableData, TableDataset
56     _name: str = field(default="")
57     conn_id: str = field(default="")
58     # Setting converter allows passing a dictionary to metadata arg
59     metadata: Metadata = field(
60         factory=Metadata,
61         converter=lambda val: Metadata(**val) if isinstance(val, dict) else val,
62     )
63     columns: list[Column] = field(factory=list)
64     temp: bool = field(default=False)
65 
66     # We need this method to pickle Table object, without this we cannot push/pull this object from xcom.
67     def __getstate__(self):
68         return self.__dict__
69 
70     def __attrs_post_init__(self) -> None:
71         if not self._name or self._name.startswith("_tmp"):
72             self.temp = True
73 
74     def _create_unique_table_name(self, prefix: str = "") -> str:
75         """
76         If a table is instantiated without a name, create a unique table for it.
77         This new name should be compatible with all supported databases.
78         """
79         schema_length = len((self.metadata and self.metadata.schema) or "") + 1
80         prefix_length = len(prefix)
81 
82         unique_id = random.choice(string.ascii_lowercase) + "".join(
83             random.choice(string.ascii_lowercase + string.digits)
84             for _ in range(MAX_TABLE_NAME_LENGTH - schema_length - prefix_length)
85         )
86         if prefix:
87             unique_id = f"{prefix}{unique_id}"
88 
89         return unique_id
90 
91     def create_similar_table(self) -> Table:
92         """
93         Create a new table with a unique name but with the same metadata.
94         """
95         return Table(  # type: ignore
96             name=self._create_unique_table_name(),
97             conn_id=self.conn_id,
98             metadata=self.metadata,
99         )
100 
101     @property
102     def sqlalchemy_metadata(self) -> MetaData:
103         """Return the Sqlalchemy metadata for the given table."""
104         if self.metadata and self.metadata.schema:
105             alchemy_metadata = MetaData(schema=self.metadata.schema)
106         else:
107             alchemy_metadata = MetaData()
108         return alchemy_metadata
109 
110     @property
111     def name(self) -> str:
112         """
113         Return either the user-defined name or auto-generate one.
114         :sphinx-autoapi-skip:
115         """
116         if self.temp and not self._name:
117             self._name = self._create_unique_table_name(TEMP_PREFIX)
118         return self._name
119 
120     @name.setter
121     def name(self, value: str) -> None:
122         """
123         Set the table name. Once this happens, the table is no longer considered temporary.
124         """
125         if not isinstance(value, property) and value != self._name:
126             self._name = value
127             self.temp = False
128 
129     @property
130     def row_count(self) -> Any:
131         """
132         Return the row count of table.
133         """
134         db = create_database(self.conn_id)
135         tb = db.get_sqla_table(table=self)
136         query = select(func.count("*")).select_from(tb)
137         return db.run_sql(query).scalar()
138 
139     def to_json(self):
140         return {
141             "class": "Table",
142             "name": self.name,
143             "metadata": {
144                 "schema": self.metadata.schema,
145                 "database": self.metadata.database,
146             },
147             "temp": self.temp,
148             "conn_id": self.conn_id,
149         }
150 
151     @classmethod
152     def from_json(cls, obj: dict):
153         return Table(
154             name=obj["name"],
155             metadata=Metadata(**obj["metadata"]),
156             temp=obj["temp"],
157             conn_id=obj["conn_id"],
158         )
159 
160     def openlineage_dataset_name(self) -> str:
161         """
162         Returns the open lineage dataset name as per
163         https://github.com/OpenLineage/OpenLineage/blob/main/spec/Naming.md
164         """
165         database = create_database(self.conn_id)
166         return database.openlineage_dataset_name(table=self)
167 
168     def openlineage_dataset_namespace(self) -> str:
169         """
170         Returns the open lineage dataset namespace as per
171         https://github.com/OpenLineage/OpenLineage/blob/main/spec/Naming.md
172         """
173         database = create_database(self.conn_id)
174         return database.openlineage_dataset_namespace()
175 
176 
177 @define(slots=False)
178 class TempTable(BaseTable):
179     """
180     Internal class to represent a Temporary table
181 
182     :sphinx-autoapi-skip:
183     """
184 
185     temp: bool = field(default=True)
186 
187 
188 @define(slots=False)
189 class Table(BaseTable, Dataset):
190     """
191     User-facing class that has information necessary to access a SQL Table. It is agnostic to the database type.
192     If no name is given, it auto-generates a name for the Table and considers it temporary.
193 
194     Temporary tables are prefixed with the prefix TEMP_PREFIX.
195 
196     :param name: The name of the database table. If name not provided then it would create a temporary name
197     :param conn_id: The Airflow connection id. This will be used to identify the right database type at the runtime
198     :param metadata: A metadata object which will have database or schema name
199     :param columns: columns which define the database table schema.
200     """
201 
202     uri: str = field(init=False)
203     extra: dict | None = field(init=False, factory=dict)
204 
205     def __new__(cls, *args, **kwargs):
206         name = kwargs.get("name") or args and args[0] or ""
207         temp = kwargs.get("temp", False)
208         if temp or (not name or name.startswith("_tmp")):
209             return TempTable(*args, **kwargs)
210         return super().__new__(cls)
211 
212     @uri.default
213     def _path_to_dataset_uri(self) -> str:
214         """Build a URI to be passed to Dataset obj introduced in Airflow 2.4"""
215         from urllib.parse import urlencode, urlparse
216 
217         path = f"astro://{self.conn_id}@"
218         db_extra = {"table": self.name}
219         if self.metadata.schema:
220             db_extra["schema"] = self.metadata.schema
221         if self.metadata.database:
222             db_extra["database"] = self.metadata.database
223         parsed_url = urlparse(url=path)
224         new_parsed_url = parsed_url._replace(query=urlencode(db_extra))
225         return new_parsed_url.geturl()
226
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/python-sdk/src/astro/table.py b/python-sdk/src/astro/table.py
--- a/python-sdk/src/astro/table.py
+++ b/python-sdk/src/astro/table.py
@@ -5,7 +5,7 @@
 from typing import Any
 
 from attr import define, field, fields_dict
-from sqlalchemy import Column, MetaData, func, select
+from sqlalchemy import Column, MetaData
 
 from astro.airflow.datasets import Dataset
 from astro.databases import create_database
@@ -132,9 +132,10 @@
         Return the row count of table.
         """
         db = create_database(self.conn_id)
-        tb = db.get_sqla_table(table=self)
-        query = select(func.count("*")).select_from(tb)
-        return db.run_sql(query).scalar()
+        result = db.run_sql(
+            f"select count(*) from {db.get_table_qualified_name(self)}"  # skipcq: BAN-B608
+        ).scalar()
+        return result
 
     def to_json(self):
         return {

{"golden_diff": "diff --git a/python-sdk/src/astro/table.py b/python-sdk/src/astro/table.py\n--- a/python-sdk/src/astro/table.py\n+++ b/python-sdk/src/astro/table.py\n@@ -5,7 +5,7 @@\n from typing import Any\n \n from attr import define, field, fields_dict\n-from sqlalchemy import Column, MetaData, func, select\n+from sqlalchemy import Column, MetaData\n \n from astro.airflow.datasets import Dataset\n from astro.databases import create_database\n@@ -132,9 +132,10 @@\n         Return the row count of table.\n         \"\"\"\n         db = create_database(self.conn_id)\n-        tb = db.get_sqla_table(table=self)\n-        query = select(func.count(\"*\")).select_from(tb)\n-        return db.run_sql(query).scalar()\n+        result = db.run_sql(\n+            f\"select count(*) from {db.get_table_qualified_name(self)}\"  # skipcq: BAN-B608\n+        ).scalar()\n+        return result\n \n     def to_json(self):\n         return {\n", "issue": "Row count not working for snowflake\n**Describe the bug**\r\nCurrently row count fails for snowflake \r\n\r\n**Expected behavior**\r\nRow count should work snowflake\r\n\r\n\n", "before_files": [{"content": "from __future__ import annotations\n\nimport random\nimport string\nfrom typing import Any\n\nfrom attr import define, field, fields_dict\nfrom sqlalchemy import Column, MetaData, func, select\n\nfrom astro.airflow.datasets import Dataset\nfrom astro.databases import create_database\n\nMAX_TABLE_NAME_LENGTH = 62\nTEMP_PREFIX = \"_tmp_\"\n\n\n@define\nclass Metadata:\n    \"\"\"\n    Contains additional information to access a SQL Table, which is very likely optional and, in some cases, may\n    be database-specific.\n\n    :param schema: A schema name\n    :param database: A database name\n    \"\"\"\n\n    # This property is used by several databases, including: Postgres, Snowflake and BigQuery (\"namespace\")\n    schema: str | None = None\n    database: str | None = None\n\n    def is_empty(self) -> bool:\n        \"\"\"Check if all the fields are None.\"\"\"\n        return all(getattr(self, field_name) is None for field_name in fields_dict(self.__class__))\n\n\n@define(slots=False)\nclass BaseTable:\n    \"\"\"\n    Base class that has information necessary to access a SQL Table. It is agnostic to the database type.\n    If no name is given, it auto-generates a name for the Table and considers it temporary.\n\n    Temporary tables are prefixed with the prefix TEMP_PREFIX.\n\n    :param name: The name of the database table. If name not provided then it would create a temporary name\n    :param conn_id: The Airflow connection id. This will be used to identify the right database type at the runtime\n    :param metadata: A metadata object which will have database or schema name\n    :param columns: columns which define the database table schema.\n    :sphinx-autoapi-skip:\n    \"\"\"\n\n    template_fields = (\"name\",)\n\n    # TODO: discuss alternative names to this class, since it contains metadata as opposed to be the\n    # SQL table itself\n    # Some ideas: TableRef, TableMetadata, TableData, TableDataset\n    _name: str = field(default=\"\")\n    conn_id: str = field(default=\"\")\n    # Setting converter allows passing a dictionary to metadata arg\n    metadata: Metadata = field(\n        factory=Metadata,\n        converter=lambda val: Metadata(**val) if isinstance(val, dict) else val,\n    )\n    columns: list[Column] = field(factory=list)\n    temp: bool = field(default=False)\n\n    # We need this method to pickle Table object, without this we cannot push/pull this object from xcom.\n    def __getstate__(self):\n        return self.__dict__\n\n    def __attrs_post_init__(self) -> None:\n        if not self._name or self._name.startswith(\"_tmp\"):\n            self.temp = True\n\n    def _create_unique_table_name(self, prefix: str = \"\") -> str:\n        \"\"\"\n        If a table is instantiated without a name, create a unique table for it.\n        This new name should be compatible with all supported databases.\n        \"\"\"\n        schema_length = len((self.metadata and self.metadata.schema) or \"\") + 1\n        prefix_length = len(prefix)\n\n        unique_id = random.choice(string.ascii_lowercase) + \"\".join(\n            random.choice(string.ascii_lowercase + string.digits)\n            for _ in range(MAX_TABLE_NAME_LENGTH - schema_length - prefix_length)\n        )\n        if prefix:\n            unique_id = f\"{prefix}{unique_id}\"\n\n        return unique_id\n\n    def create_similar_table(self) -> Table:\n        \"\"\"\n        Create a new table with a unique name but with the same metadata.\n        \"\"\"\n        return Table(  # type: ignore\n            name=self._create_unique_table_name(),\n            conn_id=self.conn_id,\n            metadata=self.metadata,\n        )\n\n    @property\n    def sqlalchemy_metadata(self) -> MetaData:\n        \"\"\"Return the Sqlalchemy metadata for the given table.\"\"\"\n        if self.metadata and self.metadata.schema:\n            alchemy_metadata = MetaData(schema=self.metadata.schema)\n        else:\n            alchemy_metadata = MetaData()\n        return alchemy_metadata\n\n    @property\n    def name(self) -> str:\n        \"\"\"\n        Return either the user-defined name or auto-generate one.\n        :sphinx-autoapi-skip:\n        \"\"\"\n        if self.temp and not self._name:\n            self._name = self._create_unique_table_name(TEMP_PREFIX)\n        return self._name\n\n    @name.setter\n    def name(self, value: str) -> None:\n        \"\"\"\n        Set the table name. Once this happens, the table is no longer considered temporary.\n        \"\"\"\n        if not isinstance(value, property) and value != self._name:\n            self._name = value\n            self.temp = False\n\n    @property\n    def row_count(self) -> Any:\n        \"\"\"\n        Return the row count of table.\n        \"\"\"\n        db = create_database(self.conn_id)\n        tb = db.get_sqla_table(table=self)\n        query = select(func.count(\"*\")).select_from(tb)\n        return db.run_sql(query).scalar()\n\n    def to_json(self):\n        return {\n            \"class\": \"Table\",\n            \"name\": self.name,\n            \"metadata\": {\n                \"schema\": self.metadata.schema,\n                \"database\": self.metadata.database,\n            },\n            \"temp\": self.temp,\n            \"conn_id\": self.conn_id,\n        }\n\n    @classmethod\n    def from_json(cls, obj: dict):\n        return Table(\n            name=obj[\"name\"],\n            metadata=Metadata(**obj[\"metadata\"]),\n            temp=obj[\"temp\"],\n            conn_id=obj[\"conn_id\"],\n        )\n\n    def openlineage_dataset_name(self) -> str:\n        \"\"\"\n        Returns the open lineage dataset name as per\n        https://github.com/OpenLineage/OpenLineage/blob/main/spec/Naming.md\n        \"\"\"\n        database = create_database(self.conn_id)\n        return database.openlineage_dataset_name(table=self)\n\n    def openlineage_dataset_namespace(self) -> str:\n        \"\"\"\n        Returns the open lineage dataset namespace as per\n        https://github.com/OpenLineage/OpenLineage/blob/main/spec/Naming.md\n        \"\"\"\n        database = create_database(self.conn_id)\n        return database.openlineage_dataset_namespace()\n\n\n@define(slots=False)\nclass TempTable(BaseTable):\n    \"\"\"\n    Internal class to represent a Temporary table\n\n    :sphinx-autoapi-skip:\n    \"\"\"\n\n    temp: bool = field(default=True)\n\n\n@define(slots=False)\nclass Table(BaseTable, Dataset):\n    \"\"\"\n    User-facing class that has information necessary to access a SQL Table. It is agnostic to the database type.\n    If no name is given, it auto-generates a name for the Table and considers it temporary.\n\n    Temporary tables are prefixed with the prefix TEMP_PREFIX.\n\n    :param name: The name of the database table. If name not provided then it would create a temporary name\n    :param conn_id: The Airflow connection id. This will be used to identify the right database type at the runtime\n    :param metadata: A metadata object which will have database or schema name\n    :param columns: columns which define the database table schema.\n    \"\"\"\n\n    uri: str = field(init=False)\n    extra: dict | None = field(init=False, factory=dict)\n\n    def __new__(cls, *args, **kwargs):\n        name = kwargs.get(\"name\") or args and args[0] or \"\"\n        temp = kwargs.get(\"temp\", False)\n        if temp or (not name or name.startswith(\"_tmp\")):\n            return TempTable(*args, **kwargs)\n        return super().__new__(cls)\n\n    @uri.default\n    def _path_to_dataset_uri(self) -> str:\n        \"\"\"Build a URI to be passed to Dataset obj introduced in Airflow 2.4\"\"\"\n        from urllib.parse import urlencode, urlparse\n\n        path = f\"astro://{self.conn_id}@\"\n        db_extra = {\"table\": self.name}\n        if self.metadata.schema:\n            db_extra[\"schema\"] = self.metadata.schema\n        if self.metadata.database:\n            db_extra[\"database\"] = self.metadata.database\n        parsed_url = urlparse(url=path)\n        new_parsed_url = parsed_url._replace(query=urlencode(db_extra))\n        return new_parsed_url.geturl()\n", "path": "python-sdk/src/astro/table.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport random\nimport string\nfrom typing import Any\n\nfrom attr import define, field, fields_dict\nfrom sqlalchemy import Column, MetaData\n\nfrom astro.airflow.datasets import Dataset\nfrom astro.databases import create_database\n\nMAX_TABLE_NAME_LENGTH = 62\nTEMP_PREFIX = \"_tmp_\"\n\n\n@define\nclass Metadata:\n    \"\"\"\n    Contains additional information to access a SQL Table, which is very likely optional and, in some cases, may\n    be database-specific.\n\n    :param schema: A schema name\n    :param database: A database name\n    \"\"\"\n\n    # This property is used by several databases, including: Postgres, Snowflake and BigQuery (\"namespace\")\n    schema: str | None = None\n    database: str | None = None\n\n    def is_empty(self) -> bool:\n        \"\"\"Check if all the fields are None.\"\"\"\n        return all(getattr(self, field_name) is None for field_name in fields_dict(self.__class__))\n\n\n@define(slots=False)\nclass BaseTable:\n    \"\"\"\n    Base class that has information necessary to access a SQL Table. It is agnostic to the database type.\n    If no name is given, it auto-generates a name for the Table and considers it temporary.\n\n    Temporary tables are prefixed with the prefix TEMP_PREFIX.\n\n    :param name: The name of the database table. If name not provided then it would create a temporary name\n    :param conn_id: The Airflow connection id. This will be used to identify the right database type at the runtime\n    :param metadata: A metadata object which will have database or schema name\n    :param columns: columns which define the database table schema.\n    :sphinx-autoapi-skip:\n    \"\"\"\n\n    template_fields = (\"name\",)\n\n    # TODO: discuss alternative names to this class, since it contains metadata as opposed to be the\n    # SQL table itself\n    # Some ideas: TableRef, TableMetadata, TableData, TableDataset\n    _name: str = field(default=\"\")\n    conn_id: str = field(default=\"\")\n    # Setting converter allows passing a dictionary to metadata arg\n    metadata: Metadata = field(\n        factory=Metadata,\n        converter=lambda val: Metadata(**val) if isinstance(val, dict) else val,\n    )\n    columns: list[Column] = field(factory=list)\n    temp: bool = field(default=False)\n\n    # We need this method to pickle Table object, without this we cannot push/pull this object from xcom.\n    def __getstate__(self):\n        return self.__dict__\n\n    def __attrs_post_init__(self) -> None:\n        if not self._name or self._name.startswith(\"_tmp\"):\n            self.temp = True\n\n    def _create_unique_table_name(self, prefix: str = \"\") -> str:\n        \"\"\"\n        If a table is instantiated without a name, create a unique table for it.\n        This new name should be compatible with all supported databases.\n        \"\"\"\n        schema_length = len((self.metadata and self.metadata.schema) or \"\") + 1\n        prefix_length = len(prefix)\n\n        unique_id = random.choice(string.ascii_lowercase) + \"\".join(\n            random.choice(string.ascii_lowercase + string.digits)\n            for _ in range(MAX_TABLE_NAME_LENGTH - schema_length - prefix_length)\n        )\n        if prefix:\n            unique_id = f\"{prefix}{unique_id}\"\n\n        return unique_id\n\n    def create_similar_table(self) -> Table:\n        \"\"\"\n        Create a new table with a unique name but with the same metadata.\n        \"\"\"\n        return Table(  # type: ignore\n            name=self._create_unique_table_name(),\n            conn_id=self.conn_id,\n            metadata=self.metadata,\n        )\n\n    @property\n    def sqlalchemy_metadata(self) -> MetaData:\n        \"\"\"Return the Sqlalchemy metadata for the given table.\"\"\"\n        if self.metadata and self.metadata.schema:\n            alchemy_metadata = MetaData(schema=self.metadata.schema)\n        else:\n            alchemy_metadata = MetaData()\n        return alchemy_metadata\n\n    @property\n    def name(self) -> str:\n        \"\"\"\n        Return either the user-defined name or auto-generate one.\n        :sphinx-autoapi-skip:\n        \"\"\"\n        if self.temp and not self._name:\n            self._name = self._create_unique_table_name(TEMP_PREFIX)\n        return self._name\n\n    @name.setter\n    def name(self, value: str) -> None:\n        \"\"\"\n        Set the table name. Once this happens, the table is no longer considered temporary.\n        \"\"\"\n        if not isinstance(value, property) and value != self._name:\n            self._name = value\n            self.temp = False\n\n    @property\n    def row_count(self) -> Any:\n        \"\"\"\n        Return the row count of table.\n        \"\"\"\n        db = create_database(self.conn_id)\n        result = db.run_sql(\n            f\"select count(*) from {db.get_table_qualified_name(self)}\"  # skipcq: BAN-B608\n        ).scalar()\n        return result\n\n    def to_json(self):\n        return {\n            \"class\": \"Table\",\n            \"name\": self.name,\n            \"metadata\": {\n                \"schema\": self.metadata.schema,\n                \"database\": self.metadata.database,\n            },\n            \"temp\": self.temp,\n            \"conn_id\": self.conn_id,\n        }\n\n    @classmethod\n    def from_json(cls, obj: dict):\n        return Table(\n            name=obj[\"name\"],\n            metadata=Metadata(**obj[\"metadata\"]),\n            temp=obj[\"temp\"],\n            conn_id=obj[\"conn_id\"],\n        )\n\n    def openlineage_dataset_name(self) -> str:\n        \"\"\"\n        Returns the open lineage dataset name as per\n        https://github.com/OpenLineage/OpenLineage/blob/main/spec/Naming.md\n        \"\"\"\n        database = create_database(self.conn_id)\n        return database.openlineage_dataset_name(table=self)\n\n    def openlineage_dataset_namespace(self) -> str:\n        \"\"\"\n        Returns the open lineage dataset namespace as per\n        https://github.com/OpenLineage/OpenLineage/blob/main/spec/Naming.md\n        \"\"\"\n        database = create_database(self.conn_id)\n        return database.openlineage_dataset_namespace()\n\n\n@define(slots=False)\nclass TempTable(BaseTable):\n    \"\"\"\n    Internal class to represent a Temporary table\n\n    :sphinx-autoapi-skip:\n    \"\"\"\n\n    temp: bool = field(default=True)\n\n\n@define(slots=False)\nclass Table(BaseTable, Dataset):\n    \"\"\"\n    User-facing class that has information necessary to access a SQL Table. It is agnostic to the database type.\n    If no name is given, it auto-generates a name for the Table and considers it temporary.\n\n    Temporary tables are prefixed with the prefix TEMP_PREFIX.\n\n    :param name: The name of the database table. If name not provided then it would create a temporary name\n    :param conn_id: The Airflow connection id. This will be used to identify the right database type at the runtime\n    :param metadata: A metadata object which will have database or schema name\n    :param columns: columns which define the database table schema.\n    \"\"\"\n\n    uri: str = field(init=False)\n    extra: dict | None = field(init=False, factory=dict)\n\n    def __new__(cls, *args, **kwargs):\n        name = kwargs.get(\"name\") or args and args[0] or \"\"\n        temp = kwargs.get(\"temp\", False)\n        if temp or (not name or name.startswith(\"_tmp\")):\n            return TempTable(*args, **kwargs)\n        return super().__new__(cls)\n\n    @uri.default\n    def _path_to_dataset_uri(self) -> str:\n        \"\"\"Build a URI to be passed to Dataset obj introduced in Airflow 2.4\"\"\"\n        from urllib.parse import urlencode, urlparse\n\n        path = f\"astro://{self.conn_id}@\"\n        db_extra = {\"table\": self.name}\n        if self.metadata.schema:\n            db_extra[\"schema\"] = self.metadata.schema\n        if self.metadata.database:\n            db_extra[\"database\"] = self.metadata.database\n        parsed_url = urlparse(url=path)\n        new_parsed_url = parsed_url._replace(query=urlencode(db_extra))\n        return new_parsed_url.geturl()\n", "path": "python-sdk/src/astro/table.py"}]}