src/pds_doi_service/core/db/doi_database.py
def write_doi_info_to_database(self, doi_record):
        """
        Write a new row to the Sqlite3 transaction database with the provided
        DOI entry information.

        Parameters
        ----------
        doi_record : DoiRecord
            The DOI record to create a database new entry with.

        Raises
        ------
        RuntimeError
            If the database transaction cannot be committed for any reason.

        """
        self.m_my_conn = self.get_connection()

        # Convert the DOI record to a dictionary representation. By doing so, we
        # can ignore database column ordering for now.
        data = dataclasses.asdict(doi_record)

        # Convert timestamps to Unix epoch floats for simpler table storage
        data["date_added"] = data["date_added"].replace(tzinfo=timezone.utc).timestamp()
        data["date_updated"] = data["date_updated"].replace(tzinfo=timezone.utc).timestamp()

        try:
            # Create and execute the query to unset the is_latest field for all
            # records with the same identifier field.
            query_string = self.query_string_for_is_latest_update(self.m_default_table_name, primary_key_column="doi")

            self.m_my_conn.execute(query_string, (doi_record.doi,))
            self.m_my_conn.commit()
        except sqlite3.Error as err:
            msg = f"Failed to update is_latest field for DOI {doi_record.doi}, reason: {err}"
            logger.error(msg)
            raise RuntimeError(msg)

        try:
            # Combine the insert and update here so the commit can be applied to both actions.
            query_string = self.query_string_for_transaction_insert(self.m_default_table_name)

            # Create the named parameters tuple in the order expected by the
            # database schema
            data_tuple = tuple([data[column] for column in self.DOI_DB_SCHEMA])

            self.m_my_conn.execute(query_string, data_tuple)
            self.m_my_conn.commit()
        except sqlite3.Error as err:
            msg = f"Failed to commit transaction for DOI {doi_record.doi}, " f"reason: {err}"
            logger.error(msg)
            raise RuntimeError(msg)

        Write a new row to the Sqlite3 transaction database with the provided
        DOI entry information.

        Parameters
        ----------
        doi_record : DoiRecord
            The DOI record to create a database new entry with.

        Raises
        ------
        RuntimeError
            If the database transaction cannot be committed for any reason.

        
Write a new row to the Sqlite3 transaction database with the provided
DOI entry information.

Parameters

doi_record : DoiRecord
The DOI record to create a database new entry with.

Raises

RuntimeError
If the database transaction cannot be committed for any reason.
Write a new row to the Sqlite3 transaction database with the provided DOI entry information . Parameters doi_record : DoiRecord The DOI record to create a database new entry with . Raises RuntimeError If the database transaction cannot be committed for any reason .
def write_doi_info_to_database(self, doi_record):
        self.m_my_conn = self.get_connection()
        data = dataclasses.asdict(doi_record)
        data["date_added"] = data["date_added"].replace(tzinfo=timezone.utc).timestamp()
        data["date_updated"] = data["date_updated"].replace(tzinfo=timezone.utc).timestamp()
        try:
            query_string = self.query_string_for_is_latest_update(self.m_default_table_name, primary_key_column="doi")
            self.m_my_conn.execute(query_string, (doi_record.doi,))
            self.m_my_conn.commit()
        except sqlite3.Error as err:
            msg = f"Failed to update is_latest field for DOI {doi_record.doi}, reason: {err}"
            logger.error(msg)
            raise RuntimeError(msg)
        try:
            query_string = self.query_string_for_transaction_insert(self.m_default_table_name)
            data_tuple = tuple([data[column] for column in self.DOI_DB_SCHEMA])
            self.m_my_conn.execute(query_string, data_tuple)
            self.m_my_conn.commit()
        except sqlite3.Error as err:
            msg = f"Failed to commit transaction for DOI {doi_record.doi}, " f"reason: {err}"
            logger.error(msg)
            raise RuntimeError(msg)
def write_doi_info_to_database ( self , doi_record ) : self . m_my_conn = self . get_connection ( ) data = dataclasses . asdict ( doi_record ) data [ "date_added" ] = data [ "date_added" ] . replace ( tzinfo = timezone . utc ) . timestamp ( ) data [ "date_updated" ] = data [ "date_updated" ] . replace ( tzinfo = timezone . utc ) . timestamp ( ) try : query_string = self . query_string_for_is_latest_update ( self . m_default_table_name , primary_key_column = "doi" ) self . m_my_conn . execute ( query_string , ( doi_record . doi , ) ) self . m_my_conn . commit ( ) except sqlite3 . Error as err : msg = f"Failed to update is_latest field for DOI {doi_record.doi}, reason: {err}" logger . error ( msg ) raise RuntimeError ( msg ) try : query_string = self . query_string_for_transaction_insert ( self . m_default_table_name ) data_tuple = tuple ( [ data [ column ] for column in self . DOI_DB_SCHEMA ] ) self . m_my_conn . execute ( query_string , data_tuple ) self . m_my_conn . commit ( ) except sqlite3 . Error as err : msg = f"Failed to commit transaction for DOI {doi_record.doi}, " f"reason: {err}" logger . error ( msg ) raise RuntimeError ( msg )
Write a new row to the Sqlite3 transaction database with the provided
DOI entry information.
Write a new row to the Sqlite3 transaction database with the provided DOI entry information .
"""
        Write a new row to the Sqlite3 transaction database with the provided
        DOI entry information.

        Parameters
        ----------
        doi_record : DoiRecord
            The DOI record to create a database new entry with.

        Raises
        ------
        RuntimeError
            If the database transaction cannot be committed for any reason.

        """ # Convert the DOI record to a dictionary representation. By doing so, we # can ignore database column ordering for now. # Convert timestamps to Unix epoch floats for simpler table storage # Create and execute the query to unset the is_latest field for all # records with the same identifier field. # Combine the insert and update here so the commit can be applied to both actions. # Create the named parameters tuple in the order expected by the # database schema
self
doi_record