File: test_db_from_dataframes.py

package info (click to toggle)
python-datacache 1.4.1-2
  • links: PTS, VCS
  • area: main
  • in suites: sid, trixie
  • size: 284 kB
  • sloc: python: 935; sh: 14; makefile: 4
file content (37 lines) | stat: -rw-r--r-- 1,374 bytes parent folder | download | duplicates (2)
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
import pandas as pd
from tempfile import NamedTemporaryFile
from datacache import db_from_dataframes, db_from_dataframe

dfA = pd.DataFrame({"numbers": [1, 2, 3], "strings": ["a", "b", "c"]})
dfB = pd.DataFrame({"wuzzles": ["nuzzle", "ruzzle"]})

def eq_(x, y):
    assert x == y

def test_database_from_dataframes():
    with NamedTemporaryFile(suffix="test.db") as f:
        db = db_from_dataframes(
            db_filename=f.name,
            dataframes={"A": dfA, "B": dfB},
            primary_keys={"A": "numbers"},
            indices={"A": [("numbers", "strings")]},
            subdir="test_datacache")
        cursor_A = db.execute("SELECT * FROM A")
        results_A = cursor_A.fetchall()
        assert(results_A == [(1, "a"), (2, "b"), (3, "c")])
        cursor_B = db.execute("SELECT * FROM B")
        results_B = cursor_B.fetchall()
        assert(results_B == [("nuzzle",), ("ruzzle",)])

def test_database_from_single_dataframe():
    with NamedTemporaryFile(suffix="test.db") as f:
        db = db_from_dataframe(
            db_filename=f.name,
            table_name="A",
            df=dfA,
            primary_key="numbers",
            indices=[("numbers", "strings")],
            subdir="test_datacache")
        cursor = db.execute("SELECT * FROM A")
        results = cursor.fetchall()
        assert(results == [(1, "a"), (2, "b"), (3, "c")])