Skip to content

medspacy.io.db_writer

DbWriter

DbWriter is a utility class for writing structured data back to a database.

Source code in medspacy/io/db_writer.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
class DbWriter:
    """DbWriter is a utility class for writing structured data back to a database."""

    def __init__(
            self,
            db_conn,
            destination_table,
            cols=None,
            col_types=None,
            doc_dtype="ents",
            create_table=False,
            drop_existing=False,
            write_batch_size=100,
    ):
        """Create a new DbWriter object.

        Args:
            db_conn: A medspacy.io.DbConnect object
            destination_table: The name of the table to write to
            cols (opt): The names of the columns of the destination table. These should align with attributes extracted
                by DocConsumer and stored in doc._.data. A set of default values can be accessed by:
                >>> DbWriter.get_default_cols()
            col_types (opt): The sql data types of the table columns. They should correspond 1:1 with cols.
                A set of default values can be accesed by:
                >>> DbWriter.get_default_col_types()
            doc_dtype: The type of data from DocConsumer to write from a doc.
                Either ("ents", "section", "context", or "doc")
            create_table (bool): Whether to create a table

        """
        self.db = db_conn
        self.destination_table = destination_table
        self._create_table = create_table
        self.drop_existing = drop_existing
        if cols is None and col_types is None:
            cols = DEFAULT_COLS[doc_dtype]
            col_types = [DEFAULT_COL_TYPES[doc_dtype][col] for col in cols]
        elif cols is None and col_types is not None:
            raise ValueError("cols must be specified if col_types is not None.")
        self.cols = cols
        self.col_types = col_types
        _validate_dtypes((doc_dtype,))
        self.doc_dtype = doc_dtype
        self.batch_size = write_batch_size

        self.insert_query = ""
        if create_table:
            self.create_table()
        self.make_insert_query()

    @classmethod
    def get_default_col_types(cls, dtypes=None):

        if dtypes is None:
            dtypes = tuple(DEFAULT_COL_TYPES.keys())
        else:
            if isinstance(dtypes, str):
                dtypes = (dtypes,)

        _validate_dtypes(dtypes)
        dtype_col_types = {
            dtype: col_types
            for (dtype, col_types) in DEFAULT_COL_TYPES.items()
            if dtype in dtypes
        }
        return dtype_col_types

    @classmethod
    def get_default_cols(cls, dtypes=None):
        if dtypes is None:
            dtypes = tuple(DEFAULT_COL_TYPES.keys())
        else:
            if isinstance(dtypes, str):
                dtypes = (dtypes,)
        _validate_dtypes(dtypes)

        dtype_cols = {
            dtype: cols
            for (dtype, cols) in DEFAULT_COL_TYPES.items()
            if dtype in dtypes
        }
        return dtype_cols

    def create_table(self):
        query = "CREATE TABLE {0} (".format(self.destination_table)
        for i, col in enumerate(self.cols):
            query += "{0} {1}".format(col, self.col_types[i])
            if i < len(self.cols) - 1:
                query += ", "
            else:
                query += ")"
        self.db.create_table(query, self.destination_table, self.drop_existing)

    def make_insert_query(self):
        col_list = ", ".join([col for col in self.cols])
        q_list = ", ".join(["?" for col in self.cols])
        self.insert_query = "INSERT INTO {0} ({1}) VALUES ({2})".format(
            self.destination_table, col_list, q_list
        )

    def write(self, docs: Union[Doc, List[Doc]]):
        """Write a list of docs or doc to a database."""
        if isinstance(docs, Doc):
            self.write_doc(docs)
        else:
            self.write_docs(docs)

    def write_doc(self, doc):
        """Write a doc to a database."""
        data = doc._.get_data(self.doc_dtype, attrs=self.cols, as_rows=True)
        self.write_data(data)

    def write_docs(self, docs, batch_size=800):
        """write a list of docs to database through bulk insert"""
        data = []
        for doc in docs:
            data.extend(doc._.get_data(self.doc_dtype, attrs=self.cols, as_rows=True))
            if len(data) >= batch_size:
                self.write_data(data)
                data = []
        if len(data) > 0:
            self.write_data(data)
        pass

    def write_data(self, data):
        self.db.write(self.insert_query, data)

    def close(self):
        self.db.close()

__init__(db_conn, destination_table, cols=None, col_types=None, doc_dtype='ents', create_table=False, drop_existing=False, write_batch_size=100)

Create a new DbWriter object.

Parameters:

Name Type Description Default
db_conn

A medspacy.io.DbConnect object

 required
destination_table

The name of the table to write to

 required
cols opt

The names of the columns of the destination table. These should align with attributes extracted by DocConsumer and stored in doc._.data. A set of default values can be accessed by:

DbWriter.get_default_cols()

 None
col_types opt

The sql data types of the table columns. They should correspond 1:1 with cols. A set of default values can be accesed by:

DbWriter.get_default_col_types()

 None
doc_dtype

The type of data from DocConsumer to write from a doc. Either ("ents", "section", "context", or "doc")

 'ents'
create_table bool

Whether to create a table

 False
Source code in medspacy/io/db_writer.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def __init__(
        self,
        db_conn,
        destination_table,
        cols=None,
        col_types=None,
        doc_dtype="ents",
        create_table=False,
        drop_existing=False,
        write_batch_size=100,
):
    """Create a new DbWriter object.

    Args:
        db_conn: A medspacy.io.DbConnect object
        destination_table: The name of the table to write to
        cols (opt): The names of the columns of the destination table. These should align with attributes extracted
            by DocConsumer and stored in doc._.data. A set of default values can be accessed by:
            >>> DbWriter.get_default_cols()
        col_types (opt): The sql data types of the table columns. They should correspond 1:1 with cols.
            A set of default values can be accesed by:
            >>> DbWriter.get_default_col_types()
        doc_dtype: The type of data from DocConsumer to write from a doc.
            Either ("ents", "section", "context", or "doc")
        create_table (bool): Whether to create a table

    """
    self.db = db_conn
    self.destination_table = destination_table
    self._create_table = create_table
    self.drop_existing = drop_existing
    if cols is None and col_types is None:
        cols = DEFAULT_COLS[doc_dtype]
        col_types = [DEFAULT_COL_TYPES[doc_dtype][col] for col in cols]
    elif cols is None and col_types is not None:
        raise ValueError("cols must be specified if col_types is not None.")
    self.cols = cols
    self.col_types = col_types
    _validate_dtypes((doc_dtype,))
    self.doc_dtype = doc_dtype
    self.batch_size = write_batch_size

    self.insert_query = ""
    if create_table:
        self.create_table()
    self.make_insert_query()

write(docs)

Write a list of docs or doc to a database.

Source code in medspacy/io/db_writer.py 
160
161
162
163
164
165
def write(self, docs: Union[Doc, List[Doc]]):
    """Write a list of docs or doc to a database."""
    if isinstance(docs, Doc):
        self.write_doc(docs)
    else:
        self.write_docs(docs)

write_doc(doc)

Write a doc to a database.

Source code in medspacy/io/db_writer.py 
167
168
169
170
def write_doc(self, doc):
    """Write a doc to a database."""
    data = doc._.get_data(self.doc_dtype, attrs=self.cols, as_rows=True)
    self.write_data(data)

write_docs(docs, batch_size=800)

write a list of docs to database through bulk insert

Source code in medspacy/io/db_writer.py 
172
173
174
175
176
177
178
179
180
181
182
def write_docs(self, docs, batch_size=800):
    """write a list of docs to database through bulk insert"""
    data = []
    for doc in docs:
        data.extend(doc._.get_data(self.doc_dtype, attrs=self.cols, as_rows=True))
        if len(data) >= batch_size:
            self.write_data(data)
            data = []
    if len(data) > 0:
        self.write_data(data)
    pass