import csv
import os.path
import subprocess
import tempfile
from onegov.core.csv import CSVFile
from typing import IO, TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Iterator
from onegov.core.csv import DefaultRow
[docs]
class DigirezDB:
""" Offers access to a Digirez Room Booking Software Database
(see http://www.digiappz.com).
Expects that mdbtools are installed (http://mdbtools.sourceforge.net).
"""
[docs]
csv_directory: tempfile.TemporaryDirectory[str] | None
def __init__(self, accessdb_path: str) -> None:
[docs]
self.accessdb_path = accessdb_path
self.csv_directory = None
@property
[docs]
def tables(self) -> list[str]:
output = subprocess.check_output(('mdb-tables', self.accessdb_path))
output_str = output.decode('utf-8').rstrip('\n ')
return output_str.split(' ')
@property
[docs]
def opened(self) -> bool:
return self.csv_directory is not None
@property
[docs]
def csv_path(self) -> str | None:
return self.csv_directory.name if self.csv_directory else None
[docs]
def open(self) -> None:
assert not self.opened
self.csv_directory = tempfile.TemporaryDirectory()
assert self.csv_path is not None
for table in self.tables:
output_path = os.path.join(self.csv_path, f'{table}.csv')
with open(output_path, 'w') as output_file:
subprocess.check_call(
args=(
'mdb-export', '-D', '%Y-%m-%dT%T',
self.accessdb_path, table
),
stdout=output_file)
@property
[docs]
def records(self) -> 'RecordsAccessor':
assert self.opened and self.csv_path
return RecordsAccessor(self.csv_path)
[docs]
class RecordsAccessor:
[docs]
files: dict[str, IO[bytes]]
def __init__(self, csv_path: str) -> None:
[docs]
self.csv_path = csv_path
self.files = {}
[docs]
def get_file(self, name: str) -> IO[bytes]:
if name not in self.files:
path = os.path.join(self.csv_path, f'{name}.csv')
self.files[name] = open(path, 'rb') # noqa: SIM115
return self.files[name]
[docs]
def __getattr__(self, name: str) -> 'Iterator[DefaultRow]':
csv_file = CSVFile(
self.get_file(name),
expected_headers=None,
dialect=csv.unix_dialect
)
return csv_file.lines