class_docstr: Database write tests that share common pytest/parametrize options.
class_docstr_tok: ["Database","write","tests","that","share","common","pytest\/parametrize","options","."]
docstr: Test append/replace ops against existing database table.
docstr_tok: ["Test","append\/replace","ops","against","existing","database","table","."]
code: def test_write_database_append_replace(
      self, engine: DbWriteEngine, tmp_path: Path
  ) -> None:
      """Test append/replace ops against existing database table."""
      df = pl.DataFrame(
          {
              "key": ["xx", "yy", "zz"],
              "value": [123, None, 789],
              "other": [5.5, 7.0, None],
          }
      )
  
      tmp_path.mkdir(exist_ok=True)
      test_db = str(tmp_path / f"test_{engine}.db")
      test_db_uri = f"sqlite:///{test_db}"
      table_name = f"test_append_{engine}"
  
      assert (
          df.write_database(
              table_name=table_name,
              connection=test_db_uri,
              engine=engine,
          )
          == 3
      )
      with pytest.raises(Exception):  # noqa: B017
          df.write_database(
              table_name=table_name,
              connection=test_db_uri,
              if_table_exists="fail",
              engine=engine,
          )
  
      assert (
          df.write_database(
              table_name=table_name,
              connection=test_db_uri,
              if_table_exists="replace",
              engine=engine,
          )
          == 3
      )
      result = pl.read_database(
          query=f"SELECT * FROM {table_name}",
          connection=create_engine(test_db_uri),
      )
      assert_frame_equal(result, df)
  
      assert (
          df[:2].write_database(
              table_name=table_name,
              connection=test_db_uri,
              if_table_exists="append",
              engine=engine,
          )
          == 2
      )
      result = pl.read_database(
          query=f"SELECT * FROM {table_name}",
          connection=create_engine(test_db_uri),
      )
      assert_frame_equal(result, pl.concat([df, df[:2]]))
  
file_path: polars/py-polars/tests/unit/io/test_database_write.py
imports: from __future__ import annotations
  import sys
  from typing import TYPE_CHECKING
  import pytest
  from sqlalchemy import create_engine
  import polars
  from polars.testing import assert_frame_equal
label_desc: Use image node_id 2 for calling the TestWriteDatabase obj's underlying member method code with example usage: obj.test_write_database_append_replace(engine, tmp_path) without return types