From 9738ac6a2b54212bf21ec96a718e0c7a85e61375 Mon Sep 17 00:00:00 2001 From: Jahnvi Thakkar Date: Wed, 25 Jun 2025 18:15:04 +0530 Subject: [PATCH 1/4] FEAT: Bulk Copy Options --- mssql_python/bcp_options.py | 168 ++++++++++++++----- mssql_python/constants.py | 15 ++ tests/test_007_bulkOptions.py | 297 ++++++++++++++++++++++++++++++++++ 3 files changed, 437 insertions(+), 43 deletions(-) create mode 100644 tests/test_007_bulkOptions.py diff --git a/mssql_python/bcp_options.py b/mssql_python/bcp_options.py index 7dab82d55..9b2f0501c 100644 --- a/mssql_python/bcp_options.py +++ b/mssql_python/bcp_options.py @@ -1,5 +1,17 @@ +""" +Copyright (c) Microsoft Corporation. +Licensed under the MIT license. +BCPOptions and ColumnFormat classes for Bulk Copy Program (BCP) operations. +""" +import logging from dataclasses import dataclass, field -from typing import List, Optional, Literal +from typing import List, Optional, Union + +# Removed unused import: BCPControlOptions + +# defining constants for BCP control options +ALLOWED_DIRECTIONS = ("in", "out", "queryout") +ALLOWED_FILE_MODES = ("native", "char", "unicode") @dataclass @@ -23,31 +35,33 @@ class ColumnFormat: Must be a positive integer. """ - prefix_len: int - data_len: int + file_col: int = 1 + user_data_type: int = 0 + prefix_len: int = 0 + data_len: int = 0 field_terminator: Optional[bytes] = None - row_terminator: Optional[bytes] = None + terminator_len: int = 0 server_col: int = 1 - file_col: int = 1 def __post_init__(self): + logging.debug("Initializing ColumnFormat: %r", self) if self.prefix_len < 0: + logging.error("prefix_len must be a non-negative integer.") raise ValueError("prefix_len must be a non-negative integer.") if self.data_len < 0: + logging.error("data_len must be a non-negative integer.") raise ValueError("data_len must be a non-negative integer.") if self.server_col <= 0: + logging.error("server_col must be a positive integer (1-based).") raise ValueError("server_col must be a positive integer (1-based).") if self.file_col <= 0: + logging.error("file_col must be a positive integer (1-based).") raise ValueError("file_col must be a positive integer (1-based).") if self.field_terminator is not None and not isinstance( self.field_terminator, bytes ): + logging.error("field_terminator must be bytes or None.") raise TypeError("field_terminator must be bytes or None.") - if self.row_terminator is not None and not isinstance( - self.row_terminator, bytes - ): - raise TypeError("row_terminator must be bytes or None.") - @dataclass class BCPOptions: @@ -71,51 +85,119 @@ class BCPOptions: columns (List[ColumnFormat]): Column formats. """ - direction: Literal["in", "out"] - data_file: str # data_file is mandatory for 'in' and 'out' + direction: str + data_file: Optional[str] = None # data_file is mandatory for 'in' and 'out' error_file: Optional[str] = None format_file: Optional[str] = None - # write_format_file is removed as 'format' direction is not actively supported + query: Optional[str] = None # For 'query' direction + bulk_mode: Optional[str] = "native" # Default to 'native' mode batch_size: Optional[int] = None max_errors: Optional[int] = None first_row: Optional[int] = None last_row: Optional[int] = None - code_page: Optional[str] = None + code_page: Optional[Union[int, str]] = None + hints: Optional[str] = None + columns: Optional[List[ColumnFormat]] = field(default_factory=list) + row_terminator: Optional[bytes] = None keep_identity: bool = False keep_nulls: bool = False - hints: Optional[str] = None - bulk_mode: Literal["native", "char", "unicode"] = "native" - columns: List[ColumnFormat] = field(default_factory=list) def __post_init__(self): - if self.direction not in ["in", "out"]: - raise ValueError("direction must be 'in' or 'out'.") + logging.debug("Initializing BCPOptions: %r", self) + if not self.direction: + logging.error("BCPOptions.direction is a required field.") + raise ValueError("BCPOptions.direction is a required field.") + + if self.direction not in ALLOWED_DIRECTIONS: + logging.error( + "BCPOptions.direction '%s' is invalid. Allowed directions are: %s.", + self.direction, ", ".join(ALLOWED_DIRECTIONS) + ) + raise ValueError( + f"BCPOptions.direction '{self.direction}' is invalid. + Allowed directions are: {', '.join(ALLOWED_DIRECTIONS)}." + ) + + if self.direction in ["in", "out"]: + if not self.data_file: + logging.error( + "BCPOptions.data_file is required for BCP direction '%s'.", + self.direction + ) + raise ValueError( + f"BCPOptions.data_file is required for BCP direction '{self.direction}'." + ) + if self.direction == "queryout" and not self.query: + logging.error("BCPOptions.query is required for BCP direction 'query'.") + raise ValueError("BCPOptions.query is required for BCP direction 'query'.") + if not self.data_file: - raise ValueError("data_file must be provided and non-empty for 'in' or 'out' directions.") - if self.error_file is None or not self.error_file: # Making error_file mandatory for in/out - raise ValueError("error_file must be provided and non-empty for 'in' or 'out' directions.") - - if self.format_file is not None and not self.format_file: - raise ValueError("format_file, if provided, must not be an empty string.") - if self.batch_size is not None and self.batch_size <= 0: - raise ValueError("batch_size must be a positive integer.") - if self.max_errors is not None and self.max_errors < 0: - raise ValueError("max_errors must be a non-negative integer.") - if self.first_row is not None and self.first_row <= 0: - raise ValueError("first_row must be a positive integer.") - if self.last_row is not None and self.last_row <= 0: - raise ValueError("last_row must be a positive integer.") - if self.last_row is not None and self.first_row is None: - raise ValueError("first_row must be specified if last_row is specified.") + logging.error( + "data_file must be provided and non-empty for 'in' or 'out' directions." + ) + raise ValueError( + "data_file must be provided and non-empty for 'in' or 'out' directions." + ) + if self.error_file is None or not self.error_file: + logging.error( + "error_file must be provided and non-empty for 'in' or 'out' directions." + ) + raise ValueError( + "error_file must be provided and non-empty for 'in' or 'out' directions." + ) + + if self.columns and self.format_file: + logging.error( + "Cannot specify both 'columns' (for bcp_colfmt) and 'format_file' " + "(for bcp_readfmt). Choose one." + ) + raise ValueError( + "Cannot specify both 'columns' (for bcp_colfmt) and 'format_file' " + "(for bcp_readfmt). Choose one." + ) + + if isinstance(self.code_page, int) and self.code_page < 0: + logging.error( + "BCPOptions.code_page, if an integer, must be non-negative." + ) + raise ValueError( + "BCPOptions.code_page, if an integer, must be non-negative." + ) + + if self.bulk_mode not in ALLOWED_FILE_MODES: + logging.error( + "BCPOptions.bulk_mode '%s' is invalid. Allowed modes are: %s.", + self.bulk_mode, ", ".join(ALLOWED_FILE_MODES) + ) + raise ValueError( + f"BCPOptions.bulk_mode '{self.bulk_mode}' is invalid. + Allowed modes are: {', '.join(ALLOWED_FILE_MODES)}." + ) + for attr_name in ["batch_size", "max_errors", "first_row", "last_row"]: + attr_value = getattr(self, attr_name) + if attr_value is not None and attr_value < 0: + logging.error( + "BCPOptions.%s must be non-negative if specified. Got %r", + attr_name, attr_value + ) + raise ValueError( + f"BCPOptions.{attr_name} must be non-negative if specified. Got {attr_value!r}" + ) + if ( self.first_row is not None and self.last_row is not None - and self.last_row < self.first_row + and self.first_row > self.last_row + ): + logging.error( + "BCPOptions.first_row cannot be greater than BCPOptions.last_row." + ) + raise ValueError( + "BCPOptions.first_row cannot be greater than BCPOptions.last_row." + ) + + if self.row_terminator is not None and not isinstance( + self.row_terminator, bytes ): - raise ValueError("last_row must be greater than or equal to first_row.") - if self.code_page is not None and not self.code_page: - raise ValueError("code_page, if provided, must not be an empty string.") - if self.hints is not None and not self.hints: - raise ValueError("hints, if provided, must not be an empty string.") - if self.bulk_mode not in ["native", "char", "unicode"]: - raise ValueError("bulk_mode must be 'native', 'char', or 'unicode'.") + logging.error("row_terminator must be bytes or None.") + raise TypeError("row_terminator must be bytes or None.") diff --git a/mssql_python/constants.py b/mssql_python/constants.py index aade503c7..055bbabe0 100644 --- a/mssql_python/constants.py +++ b/mssql_python/constants.py @@ -116,3 +116,18 @@ class ConstantsDDBC(Enum): SQL_C_WCHAR = -8 SQL_NULLABLE = 1 SQL_MAX_NUMERIC_LEN = 16 + +class BCPControlOptions(Enum): + """ + Constants for BCP control options. + The values are the string representations expected by the BCP API. + """ + BATCH_SIZE = "BCPBATCH" + MAX_ERRORS = "BCPMAXERRS" + FIRST_ROW = "BCPFIRST" + LAST_ROW = "BCPLAST" + FILE_CODE_PAGE = "BCPFILECP" + KEEP_IDENTITY = "BCPKEEPIDENTITY" + KEEP_NULLS = "BCPKEEPNULLS" + HINTS = "BCPHINTS" + SET_ROW_TERMINATOR = "BCPSETROWTERM" \ No newline at end of file diff --git a/tests/test_007_bulkOptions.py b/tests/test_007_bulkOptions.py new file mode 100644 index 000000000..15cfcaff2 --- /dev/null +++ b/tests/test_007_bulkOptions.py @@ -0,0 +1,297 @@ +"""Unit tests for mssql_python.bcp_options and BCPOptions/ColumnFormat.""" + +import os +import uuid +import pytest + +# Assuming your project structure allows these imports +from mssql_python import connect as mssql_connect # Alias to avoid conflict +from mssql_python.bcp_options import ColumnFormat, BCPOptions + +# --- Constants for Tests --- +SQL_COPT_SS_BCP = 1219 # BCP connection attribute + +# --- Database Connection Details from Environment Variables --- +DB_CONNECTION_STRING = os.getenv("DB_CONNECTION_STRING") + +# Skip all tests in this file if connection string is not provided +pytestmark = pytest.mark.skipif( + not DB_CONNECTION_STRING, + reason="DB_CONNECTION_STRING environment variable must be set for BCP integration tests.", +) + + +def get_bcp_test_conn_str(): + """Returns the connection string.""" + if not DB_CONNECTION_STRING: + # This should ideally not be reached due to pytestmark, but as a safeguard: + pytest.skip("DB_CONNECTION_STRING is not set.") + return DB_CONNECTION_STRING + + +@pytest.fixture(scope="function") +def bcp_db_setup_and_teardown(): + """ + Fixture to set up a BCP-enabled connection and a unique test table. + Yields (connection, table_name). + Cleans up the table afterwards. + """ + conn_str = get_bcp_test_conn_str() + table_name_uuid_part = str(uuid.uuid4()).replace("-", "")[:8] + table_name = f"dbo.pytest_bcp_table_{table_name_uuid_part}" + + conn = None + cursor = None + try: + conn = mssql_connect( + conn_str, attrs_before={SQL_COPT_SS_BCP: 1}, autocommit=True + ) + cursor = conn.cursor() + + cursor.execute( + f"IF OBJECT_ID('{table_name}', 'U') IS NOT NULL DROP TABLE {table_name};" + ) + cursor.execute( + f""" + CREATE TABLE {table_name} ( + id INT PRIMARY KEY, + data_col VARCHAR(255) NULL + ); + """ + ) + yield conn, table_name + finally: + if cursor: + try: + cursor.close() + except Exception as exc: + print( + f"Warning: Error closing cursor during BCP test setup/teardown: {exc}" + ) + if conn: + cursor_cleanup = None + try: + cursor_cleanup = conn.cursor() + cursor_cleanup.execute( + f"IF OBJECT_ID('{table_name}', 'U') IS NOT NULL DROP TABLE {table_name};" + ) + except Exception as exc: + print( + f"Warning: Error during BCP test cleanup (dropping table {table_name}): {exc}" + ) + finally: + if cursor_cleanup: + try: + cursor_cleanup.close() + except Exception as exc: + print(f"Warning: Error closing cleanup cursor: {exc}") + conn.close() + + +@pytest.fixture +def temp_file_pair(tmp_path): + """Provides a pair of temporary file paths for data and errors using pytest's tmp_path.""" + file_uuid_part = str(uuid.uuid4()).replace("-", "")[:8] + data_file = tmp_path / f"bcp_data_{file_uuid_part}.csv" + error_file = tmp_path / f"bcp_error_{file_uuid_part}.txt" + return data_file, error_file + + +class TestColumnFormat: + """Unit tests for the ColumnFormat class.""" + + def test_valid_instantiation_defaults(self): + """Test default instantiation of ColumnFormat.""" + cf = ColumnFormat() + assert cf.prefix_len == 0 + assert cf.data_len == 0 + assert cf.field_terminator is None + assert cf.server_col == 1 + assert cf.file_col == 1 + assert cf.user_data_type == 0 + + def test_valid_instantiation_all_params(self): + """Test instantiation of ColumnFormat with all parameters.""" + cf = ColumnFormat( + prefix_len=1, + data_len=10, + field_terminator=b",", + server_col=2, + file_col=3, + user_data_type=10, + ) + assert cf.prefix_len == 1 + assert cf.data_len == 10 + assert cf.field_terminator == b"," + assert cf.server_col == 2 + assert cf.file_col == 3 + assert cf.user_data_type == 10 + + @pytest.mark.parametrize( + "attr, value", + [ + ("prefix_len", -1), + ("data_len", -1), + ("server_col", 0), + ("server_col", -1), + ("file_col", 0), + ("file_col", -1), + ], + ) + def test_invalid_numeric_values(self, attr, value): + """Test invalid numeric values for ColumnFormat.""" + with pytest.raises(ValueError): + ColumnFormat(**{attr: value}) + + @pytest.mark.parametrize( + "attr, value", + [ + ("field_terminator", ","), + ], + ) + def test_invalid_terminator_types(self, attr, value): + """Test invalid field_terminator types for ColumnFormat.""" + with pytest.raises(TypeError): + ColumnFormat(**{attr: value}) + + +class TestBCPOptions: + """Unit tests for the BCPOptions class.""" + + _dummy_data_file = "dummy_data.csv" + _dummy_error_file = "dummy_error.txt" + + def test_valid_instantiation_in_minimal(self): + """Test minimal valid instantiation for BCPOptions (direction in).""" + opts = BCPOptions( + direction="in", + data_file=self._dummy_data_file, + error_file=self._dummy_error_file, + ) + assert opts.direction == "in" + assert opts.data_file == self._dummy_data_file + assert opts.error_file == self._dummy_error_file + assert opts.bulk_mode == "native" + + def test_valid_instantiation_out_full(self): + """Test full valid instantiation for BCPOptions (direction out).""" + cols = [ColumnFormat(file_col=1, server_col=1, field_terminator=b"\t")] + opts = BCPOptions( + direction="out", + data_file="output.csv", + error_file="errors.log", + bulk_mode="char", + batch_size=1000, + max_errors=10, + first_row=1, + last_row=100, + code_page="ACP", + hints="ORDER(id)", + columns=cols, + keep_identity=True, + keep_nulls=True, + ) + assert opts.direction == "out" + assert opts.bulk_mode == "char" + assert opts.columns == cols + assert opts.keep_identity is True + + def test_invalid_direction(self): + """Test invalid direction for BCPOptions.""" + with pytest.raises( + ValueError, match="BCPOptions.direction 'invalid_dir' is invalid" + ): + BCPOptions( + direction="invalid_dir", + data_file=self._dummy_data_file, + error_file=self._dummy_error_file, + ) + + @pytest.mark.parametrize("direction_to_test", ["in", "out"]) + def test_missing_data_file_for_in_out(self, direction_to_test): + """Test missing data_file for in/out directions in BCPOptions.""" + with pytest.raises( + ValueError, + match=f"BCPOptions.data_file is required for BCP direction '{direction_to_test}'.", + ): + BCPOptions(direction=direction_to_test, error_file=self._dummy_error_file) + + @pytest.mark.parametrize("direction_to_test", ["in", "out"]) + def test_missing_error_file_for_in_out(self, direction_to_test): + """Test missing error_file for in/out directions in BCPOptions.""" + with pytest.raises( + ValueError, + match="error_file must be provided and non-empty for 'in' or 'out' directions.", + ): + BCPOptions(direction=direction_to_test, data_file=self._dummy_data_file) + + def test_columns_and_format_file_conflict(self): + """Test conflict between columns and format_file in BCPOptions.""" + with pytest.raises( + ValueError, match="Cannot specify both 'columns' .* and 'format_file'" + ): + BCPOptions( + direction="in", + data_file=self._dummy_data_file, + error_file=self._dummy_error_file, + columns=[ColumnFormat()], + format_file="format.fmt", + ) + + def test_invalid_bulk_mode(self): + """Test invalid bulk_mode for BCPOptions.""" + with pytest.raises( + ValueError, match="BCPOptions.bulk_mode 'invalid_mode' is invalid" + ): + BCPOptions( + direction="in", + data_file=self._dummy_data_file, + error_file=self._dummy_error_file, + bulk_mode="invalid_mode", + ) + + @pytest.mark.parametrize( + "attr, value", + [ + ("batch_size", -1), + ("max_errors", -1), + ("first_row", -1), + ("last_row", -1), + ], + ) + def test_negative_control_values(self, attr, value): + """Test negative control values for BCPOptions.""" + with pytest.raises(ValueError, match=f"BCPOptions.{attr} must be non-negative"): + BCPOptions( + direction="in", + data_file=self._dummy_data_file, + error_file=self._dummy_error_file, + **{attr: value}, + ) + + def test_first_row_greater_than_last_row(self): + """Test first_row greater than last_row in BCPOptions.""" + with pytest.raises( + ValueError, + match="BCPOptions.first_row cannot be greater than BCPOptions.last_row", + ): + BCPOptions( + direction="in", + data_file=self._dummy_data_file, + error_file=self._dummy_error_file, + first_row=10, + last_row=5, + ) + + def test_invalid_codepage_negative_int(self): + """Test negative integer code_page for BCPOptions.""" + with pytest.raises( + ValueError, + match="BCPOptions.code_page, if an integer, must be non-negative", + ): + BCPOptions( + direction="in", + data_file=self._dummy_data_file, + error_file=self._dummy_error_file, + code_page=-1, + ) From 78627bc95bdd75f2d3f8a780e3a2006da874b562 Mon Sep 17 00:00:00 2001 From: Jahnvi Thakkar Date: Wed, 25 Jun 2025 18:15:22 +0530 Subject: [PATCH 2/4] FEAT: Bulk Copy Options --- mssql_python/bcp_options.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/mssql_python/bcp_options.py b/mssql_python/bcp_options.py index 9b2f0501c..0b8681714 100644 --- a/mssql_python/bcp_options.py +++ b/mssql_python/bcp_options.py @@ -114,8 +114,8 @@ def __post_init__(self): self.direction, ", ".join(ALLOWED_DIRECTIONS) ) raise ValueError( - f"BCPOptions.direction '{self.direction}' is invalid. - Allowed directions are: {', '.join(ALLOWED_DIRECTIONS)}." + f"BCPOptions.direction '{self.direction}' is invalid. " + f"Allowed directions are: {', '.join(ALLOWED_DIRECTIONS)}." ) if self.direction in ["in", "out"]: @@ -170,8 +170,8 @@ def __post_init__(self): self.bulk_mode, ", ".join(ALLOWED_FILE_MODES) ) raise ValueError( - f"BCPOptions.bulk_mode '{self.bulk_mode}' is invalid. - Allowed modes are: {', '.join(ALLOWED_FILE_MODES)}." + f"BCPOptions.bulk_mode '{self.bulk_mode}' is invalid. " + f"Allowed modes are: {', '.join(ALLOWED_FILE_MODES)}." ) for attr_name in ["batch_size", "max_errors", "first_row", "last_row"]: attr_value = getattr(self, attr_name) From ae1f289429301b19d5c43040a6e3e0d1c9c2af9b Mon Sep 17 00:00:00 2001 From: Jahnvi Thakkar Date: Mon, 30 Jun 2025 20:41:51 +0530 Subject: [PATCH 3/4] FEAT: Adding binding options --- mssql_python/__init__.py | 106 ++++++++-- mssql_python/bcp_options.py | 186 ++++++++++-------- mssql_python/constants.py | 79 +++++++- tests/test_007_bulkOptions.py | 351 ++++++++++++++++++++++++++++++---- 4 files changed, 587 insertions(+), 135 deletions(-) diff --git a/mssql_python/__init__.py b/mssql_python/__init__.py index a4f20a46e..83d5abe14 100644 --- a/mssql_python/__init__.py +++ b/mssql_python/__init__.py @@ -4,10 +4,13 @@ This module initializes the mssql_python package. """ +# Import for pooling functionality +from .pooling import PoolingManager + # Exceptions # https://www.python.org/dev/peps/pep-0249/#exceptions from .exceptions import ( - Warning, + Warning, # pylint: disable=redefined-builtin Error, InterfaceError, DatabaseError, @@ -45,25 +48,90 @@ from .logging_config import setup_logging, get_logger # Constants -from .constants import ConstantsDDBC +from .constants import ConstantsDDBC, BCPControlOptions, BCPDataTypes + +# BCP +from .bcp_options import BCPOptions, ColumnFormat # GLOBALS -# Read-Only -apilevel = "2.0" -paramstyle = "qmark" -threadsafety = 1 +# Read-Only - PEP-249 mandates these names, so we disable the invalid-name warning +apilevel = "2.0" # pylint: disable=invalid-name +paramstyle = "qmark" # pylint: disable=invalid-name +threadsafety = 1 # pylint: disable=invalid-name + +# Create direct variables for easier access to BCP data type constants Read-only +# Character/string types +SQLTEXT = BCPDataTypes.SQLTEXT.value +SQLVARCHAR = BCPDataTypes.SQLVARCHAR.value +SQLCHARACTER = BCPDataTypes.SQLCHARACTER.value +SQLBIGCHAR = BCPDataTypes.SQLBIGCHAR.value +SQLBIGVARCHAR = BCPDataTypes.SQLBIGVARCHAR.value +SQLNCHAR = BCPDataTypes.SQLNCHAR.value +SQLNVARCHAR = BCPDataTypes.SQLNVARCHAR.value +SQLNTEXT = BCPDataTypes.SQLNTEXT.value + +# Binary types +SQLBINARY = BCPDataTypes.SQLBINARY.value +SQLVARBINARY = BCPDataTypes.SQLVARBINARY.value +SQLBIGBINARY = BCPDataTypes.SQLBIGBINARY.value +SQLBIGVARBINARY = BCPDataTypes.SQLBIGVARBINARY.value +SQLIMAGE = BCPDataTypes.SQLIMAGE.value + +# Integer types +SQLBIT = BCPDataTypes.SQLBIT.value +SQLBITN = BCPDataTypes.SQLBITN.value +SQLINT1 = BCPDataTypes.SQLINT1.value +SQLINT2 = BCPDataTypes.SQLINT2.value +SQLINT4 = BCPDataTypes.SQLINT4.value +SQLINT8 = BCPDataTypes.SQLINT8.value +SQLINTN = BCPDataTypes.SQLINTN.value + +# Floating point types +SQLFLT4 = BCPDataTypes.SQLFLT4.value +SQLFLT8 = BCPDataTypes.SQLFLT8.value +SQLFLTN = BCPDataTypes.SQLFLTN.value + +# Decimal/numeric types +SQLDECIMAL = BCPDataTypes.SQLDECIMAL.value +SQLNUMERIC = BCPDataTypes.SQLNUMERIC.value +SQLDECIMALN = BCPDataTypes.SQLDECIMALN.value +SQLNUMERICN = BCPDataTypes.SQLNUMERICN.value + +# Money types +SQLMONEY = BCPDataTypes.SQLMONEY.value +SQLMONEY4 = BCPDataTypes.SQLMONEY4.value +SQLMONEYN = BCPDataTypes.SQLMONEYN.value + +# Date/time types +SQLDATETIME = BCPDataTypes.SQLDATETIME.value +SQLDATETIM4 = BCPDataTypes.SQLDATETIM4.value +SQLDATETIMN = BCPDataTypes.SQLDATETIMN.value +SQLDATEN = BCPDataTypes.SQLDATEN.value +SQLTIMEN = BCPDataTypes.SQLTIMEN.value +SQLDATETIME2N = BCPDataTypes.SQLDATETIME2N.value +SQLDATETIMEOFFSETN = BCPDataTypes.SQLDATETIMEOFFSETN.value + +# Special types +SQLUNIQUEID = BCPDataTypes.SQLUNIQUEID.value +SQLVARIANT = BCPDataTypes.SQLVARIANT.value +SQLUDT = BCPDataTypes.SQLUDT.value +SQLXML = BCPDataTypes.SQLXML.value +SQLTABLE = BCPDataTypes.SQLTABLE.value + +# BCP special values +SQL_VARLEN_DATA = BCPDataTypes.SQL_VARLEN_DATA.value +SQL_NULL_DATA = BCPDataTypes.SQL_NULL_DATA.value + -from .pooling import PoolingManager def pooling(max_size=100, idle_timeout=600): -# """ -# Enable connection pooling with the specified parameters. - -# Args: -# max_size (int): Maximum number of connections in the pool. -# idle_timeout (int): Time in seconds before idle connections are closed. - -# Returns: -# None -# """ - PoolingManager.enable(max_size, idle_timeout) - \ No newline at end of file + """ + Enable connection pooling with the specified parameters. + + Args: + max_size (int): Maximum number of connections in the pool. + idle_timeout (int): Time in seconds before idle connections are closed. + + Returns: + None + """ + PoolingManager.enable(max_size, idle_timeout) diff --git a/mssql_python/bcp_options.py b/mssql_python/bcp_options.py index 0b8681714..76dee7ead 100644 --- a/mssql_python/bcp_options.py +++ b/mssql_python/bcp_options.py @@ -1,36 +1,73 @@ """ Copyright (c) Microsoft Corporation. Licensed under the MIT license. -BCPOptions and ColumnFormat classes for Bulk Copy Program (BCP) operations. +Provides classes for configuring SQL Server Bulk Copy Program (BCP) operations. + +This module defines the core classes needed for BCP functionality: +- BindData: Represents data bindings for in-memory BCP operations +- ColumnFormat: Defines column formatting for BCP operations +- BCPOptions: Configures the overall BCP operation settings """ -import logging -from dataclasses import dataclass, field -from typing import List, Optional, Union -# Removed unused import: BCPControlOptions +from dataclasses import dataclass, field +from typing import List, Optional, Union, Any # defining constants for BCP control options ALLOWED_DIRECTIONS = ("in", "out", "queryout") ALLOWED_FILE_MODES = ("native", "char", "unicode") +@dataclass +class BindData: + """ + Represents the data binding for a column in a bulk copy operation. + Used with bcp_bind API. + + Attributes: + data: Pointer to the data to be copied. Can be primitive types or bytes. + indicator_length: Length of indicator in bytes (0, 1, 2, 4, or 8). + data_length: Count of bytes of data in the variable + (can be SQL_VARLEN_DATA/SQL_NULL_DATA). + terminator: Byte pattern marking the end of the variable, if any. + terminator_length: Count of bytes in the terminator. + data_type: The C data type of the variable (using SQL Server type tokens). + server_col: Ordinal position of the column in the database table (1-based). + """ + + data: Any = None + indicator_length: int = 0 + data_length: int = 0 # Can be SQL_VARLEN_DATA or SQL_NULL_DATA + terminator: Optional[bytes] = None + terminator_length: int = 0 + data_type: int = 0 # SQL Server data type tokens + server_col: int = 0 # 1-based column number in table + + def __post_init__(self): + if self.indicator_length not in [0, 1, 2, 4, 8]: + raise ValueError("indicator_length must be 0, 1, 2, 4, or 8.") + if self.server_col <= 0: + raise ValueError("server_col must be a positive integer (1-based).") + if self.terminator is not None and not isinstance(self.terminator, bytes): + raise TypeError("terminator must be bytes or None.") + + @dataclass class ColumnFormat: """ Represents the format of a column in a bulk copy operation. Attributes: - prefix_len (int): Option: (format_file) or (prefix_len, data_len). + prefix_len: Option: (format_file) or (prefix_len, data_len). The length of the prefix for fixed-length data types. Must be non-negative. - data_len (int): Option: (format_file) or (prefix_len, data_len). + data_len: Option: (format_file) or (prefix_len, data_len). The length of the data. Must be non-negative. - field_terminator (Optional[bytes]): Option: (-t). The field terminator string. + field_terminator: Option: (-t). The field terminator string. e.g., b',' for comma-separated values. - row_terminator (Optional[bytes]): Option: (-r). The row terminator string. + row_terminator: Option: (-r). The row terminator string. e.g., b'\\n' for newline-terminated rows. - server_col (int): Option: (format_file) or (server_col). The 1-based column number + server_col: Option: (format_file) or (server_col). The 1-based column number in the SQL Server table. Defaults to 1, representing the first column. Must be a positive integer. - file_col (int): Option: (format_file) or (file_col). The 1-based column number + file_col: Option: (format_file) or (file_col). The 1-based column number in the data file. Defaults to 1, representing the first column. Must be a positive integer. """ @@ -44,45 +81,41 @@ class ColumnFormat: server_col: int = 1 def __post_init__(self): - logging.debug("Initializing ColumnFormat: %r", self) if self.prefix_len < 0: - logging.error("prefix_len must be a non-negative integer.") raise ValueError("prefix_len must be a non-negative integer.") if self.data_len < 0: - logging.error("data_len must be a non-negative integer.") raise ValueError("data_len must be a non-negative integer.") if self.server_col <= 0: - logging.error("server_col must be a positive integer (1-based).") raise ValueError("server_col must be a positive integer (1-based).") if self.file_col <= 0: - logging.error("file_col must be a positive integer (1-based).") raise ValueError("file_col must be a positive integer (1-based).") if self.field_terminator is not None and not isinstance( self.field_terminator, bytes ): - logging.error("field_terminator must be bytes or None.") raise TypeError("field_terminator must be bytes or None.") -@dataclass + +@dataclass # pylint: disable=too-many-instance-attributes class BCPOptions: """ Represents the options for a bulk copy operation. Attributes: - direction (Literal[str]): 'in' or 'out'. Option: (-i or -o). - data_file (str): The data file. Option: (positional argument). - error_file (Optional[str]): The error file. Option: (-e). - format_file (Optional[str]): The format file to use for 'in'/'out'. Option: (-f). - batch_size (Optional[int]): The batch size. Option: (-b). - max_errors (Optional[int]): The maximum number of errors allowed. Option: (-m). - first_row (Optional[int]): The first row to process. Option: (-F). - last_row (Optional[int]): The last row to process. Option: (-L). - code_page (Optional[str]): The code page. Option: (-C). - keep_identity (bool): Keep identity values. Option: (-E). - keep_nulls (bool): Keep null values. Option: (-k). - hints (Optional[str]): Additional hints. Option: (-h). - bulk_mode (str): Bulk mode ('native', 'char', 'unicode'). Option: (-n, -c, -w). + direction: 'in' or 'out'. Option: (-i or -o). + data_file: The data file. Option: (positional argument). + error_file: The error file. Option: (-e). + format_file: The format file to use for 'in'/'out'. Option: (-f). + batch_size: The batch size. Option: (-b). + max_errors: The maximum number of errors allowed. Option: (-m). + first_row: The first row to process. Option: (-F). + last_row: The last row to process. Option: (-L). + code_page: The code page. Option: (-C). + keep_identity: Keep identity values. Option: (-E). + keep_nulls: Keep null values. Option: (-k). + hints: Additional hints. Option: (-h). + bulk_mode: Bulk mode ('native', 'char', 'unicode'). Option: (-n, -c, -w). Defaults to "native". - columns (List[ColumnFormat]): Column formats. + columns: Column formats. + bind_data: Data bindings for in-memory BCP. """ direction: str @@ -98,77 +131,75 @@ class BCPOptions: code_page: Optional[Union[int, str]] = None hints: Optional[str] = None columns: Optional[List[ColumnFormat]] = field(default_factory=list) + bind_data: Union[List[BindData], List[List[BindData]]] = field( + default_factory=list + ) # New field for bind data row_terminator: Optional[bytes] = None keep_identity: bool = False keep_nulls: bool = False + use_memory_bcp: bool = False # Flag for in-memory BCP (bind and sendrow) - def __post_init__(self): - logging.debug("Initializing BCPOptions: %r", self) + def __post_init__(self): # pylint: disable=too-many-branches if not self.direction: - logging.error("BCPOptions.direction is a required field.") raise ValueError("BCPOptions.direction is a required field.") - if self.direction not in ALLOWED_DIRECTIONS: - logging.error( - "BCPOptions.direction '%s' is invalid. Allowed directions are: %s.", - self.direction, ", ".join(ALLOWED_DIRECTIONS) + if self.bind_data and not self.use_memory_bcp: + self.use_memory_bcp = True # Automatically set if bind_data is provided + + if self.use_memory_bcp and not self.bind_data: + raise ValueError( + "BCPOptions.bind_data must be provided when use_memory_bcp is True." ) + + if self.direction not in ALLOWED_DIRECTIONS: raise ValueError( f"BCPOptions.direction '{self.direction}' is invalid. " f"Allowed directions are: {', '.join(ALLOWED_DIRECTIONS)}." ) - if self.direction in ["in", "out"]: - if not self.data_file: - logging.error( - "BCPOptions.data_file is required for BCP direction '%s'.", - self.direction + # Add this validation for in-memory BCP requiring 'in' direction + if self.use_memory_bcp and self.direction != "in": + raise ValueError("in-memory BCP operations require direction='in'") + + # Handle in-memory BCP case separately + if self.use_memory_bcp: + if not self.bind_data: + raise ValueError( + "BCPOptions.bind_data must be provided when use_memory_bcp is True." ) + # For in-memory BCP, data_file is not needed, but error_file is still useful + if not self.error_file: raise ValueError( - f"BCPOptions.data_file is required for BCP direction '{self.direction}'." + "error_file must be provided even for in-memory BCP operations." ) + else: + # Regular file-based BCP validation + if self.direction in ["in", "out"]: + if not self.data_file: + raise ValueError( + f"BCPOptions.data_file is required for file-based BCP " + f"direction '{self.direction}'." + ) + if not self.error_file: + raise ValueError( + "error_file must be provided for file-based BCP operations." + ) + if self.direction == "queryout" and not self.query: - logging.error("BCPOptions.query is required for BCP direction 'query'.") raise ValueError("BCPOptions.query is required for BCP direction 'query'.") - if not self.data_file: - logging.error( - "data_file must be provided and non-empty for 'in' or 'out' directions." - ) - raise ValueError( - "data_file must be provided and non-empty for 'in' or 'out' directions." - ) - if self.error_file is None or not self.error_file: - logging.error( - "error_file must be provided and non-empty for 'in' or 'out' directions." - ) - raise ValueError( - "error_file must be provided and non-empty for 'in' or 'out' directions." - ) - if self.columns and self.format_file: - logging.error( - "Cannot specify both 'columns' (for bcp_colfmt) and 'format_file' " - "(for bcp_readfmt). Choose one." - ) raise ValueError( "Cannot specify both 'columns' (for bcp_colfmt) and 'format_file' " "(for bcp_readfmt). Choose one." ) if isinstance(self.code_page, int) and self.code_page < 0: - logging.error( - "BCPOptions.code_page, if an integer, must be non-negative." - ) raise ValueError( "BCPOptions.code_page, if an integer, must be non-negative." ) if self.bulk_mode not in ALLOWED_FILE_MODES: - logging.error( - "BCPOptions.bulk_mode '%s' is invalid. Allowed modes are: %s.", - self.bulk_mode, ", ".join(ALLOWED_FILE_MODES) - ) raise ValueError( f"BCPOptions.bulk_mode '{self.bulk_mode}' is invalid. " f"Allowed modes are: {', '.join(ALLOWED_FILE_MODES)}." @@ -176,12 +207,9 @@ def __post_init__(self): for attr_name in ["batch_size", "max_errors", "first_row", "last_row"]: attr_value = getattr(self, attr_name) if attr_value is not None and attr_value < 0: - logging.error( - "BCPOptions.%s must be non-negative if specified. Got %r", - attr_name, attr_value - ) raise ValueError( - f"BCPOptions.{attr_name} must be non-negative if specified. Got {attr_value!r}" + f"BCPOptions.{attr_name} must be non-negative if specified. " + f"Got {attr_value}" ) if ( @@ -189,9 +217,6 @@ def __post_init__(self): and self.last_row is not None and self.first_row > self.last_row ): - logging.error( - "BCPOptions.first_row cannot be greater than BCPOptions.last_row." - ) raise ValueError( "BCPOptions.first_row cannot be greater than BCPOptions.last_row." ) @@ -199,5 +224,4 @@ def __post_init__(self): if self.row_terminator is not None and not isinstance( self.row_terminator, bytes ): - logging.error("row_terminator must be bytes or None.") raise TypeError("row_terminator must be bytes or None.") diff --git a/mssql_python/constants.py b/mssql_python/constants.py index 055bbabe0..6553a6d2d 100644 --- a/mssql_python/constants.py +++ b/mssql_python/constants.py @@ -11,6 +11,7 @@ class ConstantsDDBC(Enum): """ Constants used in the DDBC module. """ + SQL_HANDLE_ENV = 1 SQL_HANDLE_DBC = 2 SQL_HANDLE_STMT = 3 @@ -117,11 +118,13 @@ class ConstantsDDBC(Enum): SQL_NULLABLE = 1 SQL_MAX_NUMERIC_LEN = 16 + class BCPControlOptions(Enum): """ Constants for BCP control options. The values are the string representations expected by the BCP API. """ + BATCH_SIZE = "BCPBATCH" MAX_ERRORS = "BCPMAXERRS" FIRST_ROW = "BCPFIRST" @@ -130,4 +133,78 @@ class BCPControlOptions(Enum): KEEP_IDENTITY = "BCPKEEPIDENTITY" KEEP_NULLS = "BCPKEEPNULLS" HINTS = "BCPHINTS" - SET_ROW_TERMINATOR = "BCPSETROWTERM" \ No newline at end of file + SET_ROW_TERMINATOR = "BCPSETROWTERM" + + +class BCPDataTypes(Enum): + """ + SQL Server data type constants for BCP operations. + These are the native SQL Server data type tokens used with bcp_bind. + """ + + # Character/string types + SQLTEXT = 35 + SQLVARCHAR = 39 + SQLCHARACTER = 47 + SQLBIGCHAR = 175 + SQLBIGVARCHAR = 167 + SQLNCHAR = 239 + SQLNVARCHAR = 231 + SQLNTEXT = 99 + + # Binary types + SQLBINARY = 45 + SQLVARBINARY = 37 + SQLBIGBINARY = 173 + SQLBIGVARBINARY = 165 + SQLIMAGE = 34 + + # Integer types + SQLBIT = 50 + SQLBITN = 104 + SQLINT1 = 48 + SQLINT2 = 52 + SQLINT4 = 56 + SQLINT8 = 127 + SQLINTN = 38 + + # Floating point types + SQLFLT4 = 59 + SQLFLT8 = 62 + SQLFLTN = 109 + + # Decimal/numeric types + SQLDECIMAL = 106 + SQLNUMERIC = 108 + SQLDECIMALN = 106 + SQLNUMERICN = 108 + + # Money types + SQLMONEY = 60 + SQLMONEY4 = 122 + SQLMONEYN = 110 + + # Date/time types + SQLDATETIME = 61 + SQLDATETIM4 = 58 + SQLDATETIMN = 111 + SQLDATEN = 40 + SQLTIMEN = 41 + SQLDATETIME2N = 42 + SQLDATETIMEOFFSETN = 43 + + # Special types + SQLUNIQUEID = 36 + SQLVARIANT = 98 + SQLUDT = 240 + SQLXML = 241 + SQLTABLE = 243 + + # BCP special values + SQL_VARLEN_DATA = -10 + SQL_NULL_DATA = -1 + + # BCP direction codes + BCP_IN = 1 + BCP_OUT = 2 + BCP_QUERYOUT = 3 diff --git a/tests/test_007_bulkOptions.py b/tests/test_007_bulkOptions.py index 15cfcaff2..adba9afbb 100644 --- a/tests/test_007_bulkOptions.py +++ b/tests/test_007_bulkOptions.py @@ -1,12 +1,33 @@ -"""Unit tests for mssql_python.bcp_options and BCPOptions/ColumnFormat.""" - +""" +Tests for bulk options functionality in the mssql-python package. +This module tests BCPOptions, ColumnFormat, BindData classes and their behavior. +""" import os import uuid import pytest -# Assuming your project structure allows these imports -from mssql_python import connect as mssql_connect # Alias to avoid conflict -from mssql_python.bcp_options import ColumnFormat, BCPOptions +try: + from mssql_python import connect as mssql_connect # Alias to avoid conflict + from mssql_python.bcp_options import ColumnFormat, BCPOptions, BindData + from mssql_python import ( + SQLINT4, + SQLVARCHAR, + SQLNVARCHAR, + SQL_VARLEN_DATA, + SQL_NULL_DATA, + ) +except ImportError: + # Mock imports for when the package is not installed + # This allows the test file to be loaded without errors by static analysis tools + mssql_connect = None + ColumnFormat = None + BCPOptions = None + BindData = None + SQLINT4 = None + SQLVARCHAR = None + SQLNVARCHAR = None + SQL_VARLEN_DATA = None + SQL_NULL_DATA = None # --- Constants for Tests --- SQL_COPT_SS_BCP = 1219 # BCP connection attribute @@ -64,9 +85,9 @@ def bcp_db_setup_and_teardown(): if cursor: try: cursor.close() - except Exception as exc: + except Exception as e: # pylint: disable=broad-except print( - f"Warning: Error closing cursor during BCP test setup/teardown: {exc}" + f"Warning: Error closing cursor during BCP test setup/teardown: {e}" ) if conn: cursor_cleanup = None @@ -75,16 +96,16 @@ def bcp_db_setup_and_teardown(): cursor_cleanup.execute( f"IF OBJECT_ID('{table_name}', 'U') IS NOT NULL DROP TABLE {table_name};" ) - except Exception as exc: + except Exception as e: # pylint: disable=broad-except print( - f"Warning: Error during BCP test cleanup (dropping table {table_name}): {exc}" + f"Warning: Error during BCP test cleanup (dropping table {table_name}): {e}" ) finally: if cursor_cleanup: try: cursor_cleanup.close() - except Exception as exc: - print(f"Warning: Error closing cleanup cursor: {exc}") + except Exception as e: # pylint: disable=broad-except + print(f"Warning: Error closing cleanup cursor: {e}") conn.close() @@ -97,11 +118,12 @@ def temp_file_pair(tmp_path): return data_file, error_file +# --- Tests for bcp_options.py (Unit tests, no mocking needed) --- class TestColumnFormat: - """Unit tests for the ColumnFormat class.""" + """Test class for the ColumnFormat class which validates column formats for BCP operations.""" def test_valid_instantiation_defaults(self): - """Test default instantiation of ColumnFormat.""" + """Test that ColumnFormat initializes with correct default values.""" cf = ColumnFormat() assert cf.prefix_len == 0 assert cf.data_len == 0 @@ -111,7 +133,7 @@ def test_valid_instantiation_defaults(self): assert cf.user_data_type == 0 def test_valid_instantiation_all_params(self): - """Test instantiation of ColumnFormat with all parameters.""" + """Test that ColumnFormat initializes correctly with all parameters specified.""" cf = ColumnFormat( prefix_len=1, data_len=10, @@ -139,7 +161,7 @@ def test_valid_instantiation_all_params(self): ], ) def test_invalid_numeric_values(self, attr, value): - """Test invalid numeric values for ColumnFormat.""" + """Test that ColumnFormat rejects invalid numeric values.""" with pytest.raises(ValueError): ColumnFormat(**{attr: value}) @@ -150,19 +172,19 @@ def test_invalid_numeric_values(self, attr, value): ], ) def test_invalid_terminator_types(self, attr, value): - """Test invalid field_terminator types for ColumnFormat.""" + """Test that ColumnFormat rejects invalid terminator types.""" with pytest.raises(TypeError): ColumnFormat(**{attr: value}) class TestBCPOptions: - """Unit tests for the BCPOptions class.""" + """Test class for BCPOptions which configures bulk copy operations.""" _dummy_data_file = "dummy_data.csv" _dummy_error_file = "dummy_error.txt" def test_valid_instantiation_in_minimal(self): - """Test minimal valid instantiation for BCPOptions (direction in).""" + """Test minimal valid instantiation for 'in' direction.""" opts = BCPOptions( direction="in", data_file=self._dummy_data_file, @@ -174,7 +196,7 @@ def test_valid_instantiation_in_minimal(self): assert opts.bulk_mode == "native" def test_valid_instantiation_out_full(self): - """Test full valid instantiation for BCPOptions (direction out).""" + """Test full valid instantiation for 'out' direction with all parameters.""" cols = [ColumnFormat(file_col=1, server_col=1, field_terminator=b"\t")] opts = BCPOptions( direction="out", @@ -197,7 +219,7 @@ def test_valid_instantiation_out_full(self): assert opts.keep_identity is True def test_invalid_direction(self): - """Test invalid direction for BCPOptions.""" + """Test rejection of invalid direction values.""" with pytest.raises( ValueError, match="BCPOptions.direction 'invalid_dir' is invalid" ): @@ -207,26 +229,51 @@ def test_invalid_direction(self): error_file=self._dummy_error_file, ) - @pytest.mark.parametrize("direction_to_test", ["in", "out"]) - def test_missing_data_file_for_in_out(self, direction_to_test): - """Test missing data_file for in/out directions in BCPOptions.""" + @pytest.mark.parametrize("direction_to_test", ["out"]) + def test_missing_data_file_for_out(self, direction_to_test): + """Test that data_file is required for 'out' direction.""" with pytest.raises( ValueError, - match=f"BCPOptions.data_file is required for BCP direction '{direction_to_test}'.", + match="BCPOptions.data_file is required for file-based BCP direction" ): BCPOptions(direction=direction_to_test, error_file=self._dummy_error_file) - @pytest.mark.parametrize("direction_to_test", ["in", "out"]) - def test_missing_error_file_for_in_out(self, direction_to_test): - """Test missing error_file for in/out directions in BCPOptions.""" + def test_missing_data_file_for_in(self): + """Test that data_file is required for 'in' direction when not using memory BCP.""" with pytest.raises( ValueError, - match="error_file must be provided and non-empty for 'in' or 'out' directions.", + match="BCPOptions.data_file is required for file-based BCP direction 'in'" ): - BCPOptions(direction=direction_to_test, data_file=self._dummy_data_file) + BCPOptions( + direction="in", error_file=self._dummy_error_file, use_memory_bcp=False + ) + + @pytest.mark.parametrize("direction_to_test", ["in", "out"]) + def test_missing_error_file_for_any_direction(self, direction_to_test): + """Test that error_file is required for all directions.""" + if direction_to_test == "in": + with pytest.raises( + ValueError, + match="error_file must be provided even for in-memory BCP operations", + ): + BCPOptions( + direction=direction_to_test, + use_memory_bcp=True, + bind_data=[ + BindData( + data=123, data_type=SQLINT4, data_length=4, server_col=1 + ) + ], + ) + else: + with pytest.raises( + ValueError, + match="error_file must be provided for file-based BCP operations", + ): + BCPOptions(direction=direction_to_test, data_file=self._dummy_data_file) def test_columns_and_format_file_conflict(self): - """Test conflict between columns and format_file in BCPOptions.""" + """Test that columns and format_file parameters cannot be used together.""" with pytest.raises( ValueError, match="Cannot specify both 'columns' .* and 'format_file'" ): @@ -239,7 +286,7 @@ def test_columns_and_format_file_conflict(self): ) def test_invalid_bulk_mode(self): - """Test invalid bulk_mode for BCPOptions.""" + """Test rejection of invalid bulk_mode values.""" with pytest.raises( ValueError, match="BCPOptions.bulk_mode 'invalid_mode' is invalid" ): @@ -260,7 +307,7 @@ def test_invalid_bulk_mode(self): ], ) def test_negative_control_values(self, attr, value): - """Test negative control values for BCPOptions.""" + """Test rejection of negative control values.""" with pytest.raises(ValueError, match=f"BCPOptions.{attr} must be non-negative"): BCPOptions( direction="in", @@ -270,7 +317,7 @@ def test_negative_control_values(self, attr, value): ) def test_first_row_greater_than_last_row(self): - """Test first_row greater than last_row in BCPOptions.""" + """Test rejection when first_row > last_row.""" with pytest.raises( ValueError, match="BCPOptions.first_row cannot be greater than BCPOptions.last_row", @@ -284,7 +331,7 @@ def test_first_row_greater_than_last_row(self): ) def test_invalid_codepage_negative_int(self): - """Test negative integer code_page for BCPOptions.""" + """Test rejection of negative code_page values.""" with pytest.raises( ValueError, match="BCPOptions.code_page, if an integer, must be non-negative", @@ -295,3 +342,239 @@ def test_invalid_codepage_negative_int(self): error_file=self._dummy_error_file, code_page=-1, ) + + def test_valid_memory_bcp_with_bind_data(self): + """Test that in-memory BCP with bind data is properly configured.""" + bind_data_item = BindData( + data=123, data_type=SQLINT4, data_length=4, server_col=1 + ) + + opts = BCPOptions( + direction="in", + use_memory_bcp=True, + bind_data=[bind_data_item], + error_file=self._dummy_error_file, + ) + + assert opts.direction == "in" + assert opts.use_memory_bcp is True + assert len(opts.bind_data) == 1 + assert opts.bind_data[0].data == 123 + assert opts.bind_data[0].data_type == SQLINT4 + assert opts.data_file is None # Data file should be None for memory BCP + + def test_valid_memory_bcp_with_multiple_rows(self): + """Test that multi-row binding is properly configured.""" + # Define two rows with two columns each + row1 = [ + BindData(data=1001, data_type=SQLINT4, data_length=4, server_col=1), + BindData( + data="Row 1 Data", + data_type=SQLVARCHAR, + data_length=SQL_VARLEN_DATA, + terminator=b"\0", + terminator_length=1, + server_col=2, + ), + ] + + row2 = [ + BindData(data=1002, data_type=SQLINT4, data_length=4, server_col=1), + BindData( + data="Row 2 Data", + data_type=SQLVARCHAR, + data_length=SQL_VARLEN_DATA, + terminator=b"\0", + terminator_length=1, + server_col=2, + ), + ] + + opts = BCPOptions( + direction="in", + use_memory_bcp=True, + bind_data=[row1, row2], # List of rows, where each row is a list of BindData + error_file=self._dummy_error_file, + ) + + assert opts.direction == "in" + assert opts.use_memory_bcp is True + assert len(opts.bind_data) == 2 # Two rows + assert len(opts.bind_data[0]) == 2 # First row has two columns + assert opts.bind_data[0][0].data == 1001 # First column of first row + assert opts.bind_data[1][0].data == 1002 # First column of second row + assert opts.bind_data[0][1].data == "Row 1 Data" # Second column of first row + + def test_memory_bcp_requires_in_direction(self): + """Test that memory BCP requires 'in' direction.""" + with pytest.raises( + ValueError, match="in-memory BCP operations require direction='in'" + ): + BCPOptions( + direction="out", + use_memory_bcp=True, + bind_data=[ + BindData(data=123, data_type=SQLINT4, data_length=4, server_col=1) + ], + error_file=self._dummy_error_file, + ) + + def test_memory_bcp_requires_bind_data(self): + """Test that memory BCP requires bind_data.""" + with pytest.raises( + ValueError, + match="BCPOptions.bind_data must be provided when use_memory_bcp is True", + ): + BCPOptions( + direction="in", use_memory_bcp=True, error_file=self._dummy_error_file + ) + + def test_bind_data_requires_memory_bcp(self): + """Test that binding data automatically enables use_memory_bcp.""" + opts = BCPOptions( + direction="in", + bind_data=[ + BindData(data=123, data_type=SQLINT4, data_length=4, server_col=1) + ], + error_file=self._dummy_error_file, + ) + assert opts.use_memory_bcp is True + + def test_memory_bcp_doesnt_require_data_file(self): + """Test that memory BCP doesn't require data_file.""" + opts = BCPOptions( + direction="in", + use_memory_bcp=True, + bind_data=[ + BindData(data=123, data_type=SQLINT4, data_length=4, server_col=1) + ], + error_file=self._dummy_error_file, + ) + assert opts.data_file is None # Data file should be None for memory BCP + + def test_bind_data_with_null_values(self): + """Test that NULL values are properly configured in bind data.""" + bind_data_item = BindData( + data=None, + data_type=SQLINT4, + indicator_length=4, + data_length=SQL_NULL_DATA, + server_col=1, + ) + + opts = BCPOptions( + direction="in", + use_memory_bcp=True, + bind_data=[bind_data_item], + error_file=self._dummy_error_file, + ) + + assert opts.bind_data[0].data is None + assert opts.bind_data[0].indicator_length == 4 + assert opts.bind_data[0].data_length == SQL_NULL_DATA + + +class TestBindData: + """Test class for BindData which defines column data for memory-based bulk operations.""" + + def test_valid_instantiation_defaults(self): + """Test valid instantiation with minimal parameters.""" + bind_data = BindData(data=123, data_type=SQLINT4, data_length=4, server_col=1) + assert bind_data.data == 123 + assert bind_data.data_type == SQLINT4 + assert bind_data.data_length == 4 + assert bind_data.server_col == 1 + assert bind_data.indicator_length == 0 + assert bind_data.terminator is None + assert bind_data.terminator_length == 0 + + def test_valid_instantiation_all_params(self): + """Test valid instantiation with all parameters.""" + bind_data = BindData( + data="test", + data_type=SQLVARCHAR, + indicator_length=0, + data_length=SQL_VARLEN_DATA, + terminator=b"\0", + terminator_length=1, + server_col=2, + ) + assert bind_data.data == "test" + assert bind_data.data_type == SQLVARCHAR + assert bind_data.indicator_length == 0 + assert bind_data.data_length == SQL_VARLEN_DATA + assert bind_data.terminator == b"\0" + assert bind_data.terminator_length == 1 + assert bind_data.server_col == 2 + + def test_null_data_requires_sql_null_data(self): + """Test NULL data configuration with SQL_NULL_DATA.""" + bind_data = BindData( + data=None, + data_type=SQLINT4, + indicator_length=4, + data_length=SQL_NULL_DATA, + server_col=1, + ) + assert bind_data.data is None + assert bind_data.data_length == SQL_NULL_DATA + + def test_sql_null_data_requires_null_data(self): + """Test SQL_NULL_DATA with NULL data configuration.""" + bind_data = BindData( + data=None, + data_type=SQLINT4, + indicator_length=4, + data_length=SQL_NULL_DATA, + server_col=1, + ) + assert bind_data.data is None + assert bind_data.data_length == SQL_NULL_DATA + + def test_null_data_requires_indicator(self): + """Test NULL data with indicator length configuration.""" + bind_data = BindData( + data=None, + data_type=SQLINT4, + indicator_length=4, # Valid indicator length + data_length=SQL_NULL_DATA, + server_col=1, + ) + assert bind_data.indicator_length == 4 + + def test_invalid_server_col(self): + """Test that server_col must be positive.""" + with pytest.raises(ValueError, match="server_col must be a positive integer"): + BindData( + data=123, + data_type=SQLINT4, + data_length=4, + server_col=0, # Should be > 0 + ) + + def test_varlen_data_requires_terminator(self): + """Test variable length data with terminator configuration.""" + bind_data = BindData( + data="test", + data_type=SQLVARCHAR, + data_length=SQL_VARLEN_DATA, + terminator=b"\0", + terminator_length=1, + server_col=1, + ) + assert bind_data.terminator == b"\0" + assert bind_data.data_length == SQL_VARLEN_DATA + + def test_unicode_string_with_nvarchar(self): + """Test that Unicode strings work with NVARCHAR.""" + unicode_text = "Unicode 文字" + bind_data = BindData( + data=unicode_text, + data_type=SQLNVARCHAR, + data_length=SQL_VARLEN_DATA, + terminator=b"\0", + terminator_length=1, + server_col=1, + ) + assert bind_data.data == unicode_text + assert bind_data.data_type == SQLNVARCHAR From e312279ccb96bd4700020654b490e336cd7a7178 Mon Sep 17 00:00:00 2001 From: Jahnvi Thakkar Date: Mon, 30 Jun 2025 20:57:55 +0530 Subject: [PATCH 4/4] FEAT: Removing pylint comments --- mssql_python/__init__.py | 10 +++++----- mssql_python/bcp_options.py | 4 ++-- tests/test_007_bulkOptions.py | 6 +++--- 3 files changed, 10 insertions(+), 10 deletions(-) diff --git a/mssql_python/__init__.py b/mssql_python/__init__.py index 83d5abe14..62c65eb41 100644 --- a/mssql_python/__init__.py +++ b/mssql_python/__init__.py @@ -10,7 +10,7 @@ # Exceptions # https://www.python.org/dev/peps/pep-0249/#exceptions from .exceptions import ( - Warning, # pylint: disable=redefined-builtin + Warning, Error, InterfaceError, DatabaseError, @@ -54,10 +54,10 @@ from .bcp_options import BCPOptions, ColumnFormat # GLOBALS -# Read-Only - PEP-249 mandates these names, so we disable the invalid-name warning -apilevel = "2.0" # pylint: disable=invalid-name -paramstyle = "qmark" # pylint: disable=invalid-name -threadsafety = 1 # pylint: disable=invalid-name +# Read-Only - PEP-249 mandates these names +apilevel = "2.0" +paramstyle = "qmark" +threadsafety = 1 # Create direct variables for easier access to BCP data type constants Read-only # Character/string types diff --git a/mssql_python/bcp_options.py b/mssql_python/bcp_options.py index 76dee7ead..d7ef435ef 100644 --- a/mssql_python/bcp_options.py +++ b/mssql_python/bcp_options.py @@ -95,7 +95,7 @@ def __post_init__(self): raise TypeError("field_terminator must be bytes or None.") -@dataclass # pylint: disable=too-many-instance-attributes +@dataclass class BCPOptions: """ Represents the options for a bulk copy operation. @@ -139,7 +139,7 @@ class BCPOptions: keep_nulls: bool = False use_memory_bcp: bool = False # Flag for in-memory BCP (bind and sendrow) - def __post_init__(self): # pylint: disable=too-many-branches + def __post_init__(self): if not self.direction: raise ValueError("BCPOptions.direction is a required field.") diff --git a/tests/test_007_bulkOptions.py b/tests/test_007_bulkOptions.py index adba9afbb..4ca9b0906 100644 --- a/tests/test_007_bulkOptions.py +++ b/tests/test_007_bulkOptions.py @@ -85,7 +85,7 @@ def bcp_db_setup_and_teardown(): if cursor: try: cursor.close() - except Exception as e: # pylint: disable=broad-except + except Exception as e: print( f"Warning: Error closing cursor during BCP test setup/teardown: {e}" ) @@ -96,7 +96,7 @@ def bcp_db_setup_and_teardown(): cursor_cleanup.execute( f"IF OBJECT_ID('{table_name}', 'U') IS NOT NULL DROP TABLE {table_name};" ) - except Exception as e: # pylint: disable=broad-except + except Exception as e: print( f"Warning: Error during BCP test cleanup (dropping table {table_name}): {e}" ) @@ -104,7 +104,7 @@ def bcp_db_setup_and_teardown(): if cursor_cleanup: try: cursor_cleanup.close() - except Exception as e: # pylint: disable=broad-except + except Exception as e: print(f"Warning: Error closing cleanup cursor: {e}") conn.close()