eldoc/mods/db.py

import os
from peewee import Model, CharField, DateTimeField, TextField, ForeignKeyField, DateField
from playhouse.sqlite_ext import CSqliteExtDatabase, FTS5Model, AutoIncrementField, SearchField, RowIDField, JSONField

var_dir = os.path.join(os.path.abspath(os.environ['HOME']), '.eldoc')
db_dir = os.path.join(var_dir, 'db')
db_file = os.path.join(db_dir, 'eldoc.sqlite3')

if not os.path.exists(var_dir):
    os.mkdir(var_dir)
if not os.path.exists(db_dir):
    os.mkdir(db_dir)

class DBO:
    def __init__(self):
        self.db = CSqliteExtDatabase(db_file, pragmas={
            'journal_mode': 'wal',
            'cache_size': -1 * 64000,  # 64MB
            'foreign_keys': 'on',
            'ignore_check_constraints': 'off',
            'synchronous': 'off'})
        class BaseModel(Model):
            'Базовая таблица'
            class Meta:
                database = self.db
        class BaseFTSModel(FTS5Model):
            class Meta:
                database = self.db
                options = {'tokenize': 'porter'}
        class Patient(BaseModel):
            id = AutoIncrementField()
            last_name = TextField()
            first_name = TextField()
            middle_name = TextField(null=True)
            birth_date = DateField()
            gender = CharField(6)
            doc_type = CharField(32, null=True)
            doc_serial = TextField(null=True)
            doc_number = TextField(null=True)
            policy_number = TextField(null=True)
            policy_company = TextField(null=True)
            snils_number = TextField(null=True)
            notes = TextField(null=True)
        class Catalog(BaseModel):
            id = AutoIncrementField()
            system_id = TextField(null=True)
            name = TextField()
        class CatalogRecord(BaseModel):
            id = AutoIncrementField()
            catalog = ForeignKeyField(Catalog, backref='catalog_records', on_delete='CASCADE')
            text = TextField()
        class CatalogRecordIndex(BaseFTSModel):
            rowid = RowIDField()
            text = SearchField()
        class Reception(BaseModel):
            id = AutoIncrementField()
            patient = ForeignKeyField(Patient, backref='receptions', on_delete='CASCADE')
            time = DateTimeField()
        class ReceptionDiagnosis(BaseModel):
            id = AutoIncrementField()
            reception = ForeignKeyField(Reception, backref='reception_diagnosisses', on_delete='CASCADE')
            diagnosis = ForeignKeyField(CatalogRecord, backref='reception_diagnosisses', on_delete='CASCADE')
        class ReceptionAnamnesis(BaseModel):
            id = AutoIncrementField()
            reception = ForeignKeyField(Reception, backref='reception_anamnesisses', on_delete='CASCADE')
            text = TextField()
        class PatientIndex(BaseFTSModel):
            rowid = RowIDField()
            fio = SearchField()
        ### Настройки
        class Settings(BaseModel):
            key = TextField()
            val = JSONField()
        ###
        self._Patient = Patient
        self._Catalog = Catalog
        self._CatalogRecord = CatalogRecord
        self._CatalogRecordIndex = CatalogRecordIndex
        self._Reception = Reception
        self._ReceptionDiagnosis = ReceptionDiagnosis
        self._ReceptionAnamnesis = ReceptionAnamnesis
        self._PatientIndex = PatientIndex
        self._Settings = Settings
        self.db.connect()
    def create(self):
        self._Patient.add_index(
            self._Patient.index(self._Patient.last_name, self._Patient.first_name, self._Patient.birth_date, unique=True)
            )
        self.db.create_tables([
            self._Patient,
            self._PatientIndex,
            self._Catalog,
            self._CatalogRecord,
            self._CatalogRecordIndex,
            self._Reception,
            self._ReceptionDiagnosis,
            self._ReceptionAnamnesis,
            self._Settings
            ])

def search_patients(q):
    with dbo.db.atomic():
        return (dbo._Patient.select()
                .join(
                    dbo._PatientIndex,
                    on=(dbo._Patient.id == dbo._PatientIndex.rowid))
                .where(dbo._PatientIndex.match(q))
                .order_by(dbo._PatientIndex.bm25()))
def store_patient_index(pat):
    fio_list = [pat.last_name, pat.first_name]
    if pat.middle_name:
        fio_list.append(pat.middle_name)
    dbo._PatientIndex.insert(
        {
            dbo._PatientIndex.rowid: pat.id,
            dbo._PatientIndex.fio: ' '.join(fio_list)
            }
        ).execute()
def update_patient_index(pat):
    fio_list = [pat.last_name, pat.first_name]
    if pat.middle_name:
        fio_list.append(pat.middle_name)
    dbo._PatientIndex.update(
        {
            dbo._PatientIndex.fio: ' '.join(fio_list)
            }
        ).where(dbo._PatientIndex.rowid == pat.id).execute()


dbo = DBO()
dbo.create()