Miscellaneous changes

Author: hunyadi

pysqlsync/connection.py

@@ -96,5 +96,5 @@ def __str__(self) -> str:
  port = f":{self.port}" if self.port else ""
  username = f"{quote(self.username, safe='')}@" if self.username else ""
  database = f"/{quote(self.database, safe='')}" if self.database else ""
- ssl = f"?ssl={self.ssl}" if self.ssl else ""
+ ssl = f"?ssl={self.ssl.value}" if self.ssl else ""
  return f"{username}{host}{port}{database}{ssl}"

pysqlsync/dialect/mssql/discovery.py

@@ -47,9 +47,21 @@ async def get_columns(self, table_id: SupportsQualifiedId) -> list[Column]:
  " INNER JOIN sys.columns AS c ON c.object_id = t.object_id\n"
  f"WHERE s.name = {quote(table_id.scope_id or 'dbo')} AND t.name = {quote(table_id.local_id)} AND c.is_identity = 1;",
  )
+ default_columns = await self.conn.query_all(
+ tuple[str, str],
+ "SELECT c.name, d.definition\n"
+ "FROM sys.schemas AS s\n"
+ " INNER JOIN sys.tables AS t ON t.schema_id = s.schema_id\n"
+ " INNER JOIN sys.columns AS c ON c.object_id = t.object_id\n"
+ " INNER JOIN sys.default_constraints AS d ON d.object_id = c.default_object_id\n"
+ f"WHERE s.name = {quote(table_id.scope_id or 'dbo')} AND t.name = {quote(table_id.local_id)} AND c.default_object_id != 0;",
+ )
+ default_values = dict(default_columns)
  for c in columns:
  if c.name.id in identity_columns:
  c.identity = True
+ if c.name.id in default_values:
+ c.default = default_values[c.name.id]
 
  return columns
 

setup.cfg

@@ -29,9 +29,9 @@ include_package_data = True
 packages = find:
 python_requires = >=3.9
 install_requires =
- certifi >= 2024.8.30
- truststore >= 0.9; python_version>="3.10"
  json_strong_typing >= 0.3.2
+ certifi >= 2024.8.30; python_version<"3.10"
+ truststore >= 0.9; python_version>="3.10"
  typing_extensions >= 4.8; python_version<"3.12"
 
 [options.extras_require]

tests/test_api.py

@@ -46,6 +46,7 @@ def test_connection_query_parameters(self) -> None:
  self.assertEqual(params.port, port)
  self.assertEqual(params.database, database)
  self.assertEqual(params.ssl, None)
+ self.assertEqual(str(params), r"server.example.com:2310/database")
 
  url = f"{url_prefix}?ssl=verify-full"
  dialect, params = get_parameters(url)
@@ -54,6 +55,9 @@ def test_connection_query_parameters(self) -> None:
  self.assertEqual(params.port, port)
  self.assertEqual(params.database, database)
  self.assertEqual(params.ssl, ConnectionSSLMode.verify_full)
+ self.assertEqual(
+ str(params), r"server.example.com:2310/database?ssl=verify-full"
+ )
 
 
 if __name__ == "__main__":