c1019695-ebf0-419c-9312-e0b247a62c6a

Committed 21 May 2025 04:27PM UTC coverage: 99.611% (-0.4%) from 100.0%

Build # c1019695-ebf0-419c-9312-e0b247a62c6a

Build Type

Pull #24

circleci

Committed by

maxkadel

Commit Message

Linting fixes

Pull Request Pull Request #24: Green locally - connect to Postgres DB

Run Details

204 of 207 new or added lines in 9 files covered. (98.55%)

2 existing lines in 1 file now uncovered.

769 of 772 relevant lines covered (99.61%)

1.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.14

/src/marc_to_db.py

import os.path
import time
import psycopg2
from pymarc import map_xml
from config import settings
from src.marc_record import MarcRecord
from src.streaming_json_handler import map_json

CREATE_TABLE_SQL = """CREATE TABLE IF NOT EXISTS records (
id TEXT,
title TEXT,
author TEXT,
publication_year TEXT,
pagination TEXT,
edition TEXT,
publisher_name TEXT,
type_of VARCHAR,
is_electronic_resource BOOL,
source_file TEXT,
UNIQUE (id)
);
"""

CREATE_RECORD_SQL = """INSERT INTO records VALUES
(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""


class MarcToDb:
    conn = psycopg2.connect(
        database=settings.db_name,
        user=settings.db_user,
        host=settings.db_host,
        port=settings.db_port,
    )
    conn.autocommit = True

    @classmethod
    def find_or_create_table(cls):
        with cls.conn.cursor() as cur:
            cur.execute(CREATE_TABLE_SQL)

    def __init__(self, input_file_path):
        self.input_file_path = input_file_path
        self.conn = MarcToDb.conn
        self.source_file, self.file_extension = os.path.splitext(
            os.path.basename(self.input_file_path)
        )
        self.cursor = self.conn.cursor()

    def to_db(self):
        print(
            f"""time: {time.asctime(time.localtime())} -
                writing records in {self.input_file_path} to database
            """
        )
        if self.file_extension == ".xml":
            map_xml(self.add_record, self.input_file_path)
        elif self.file_extension == ".json":
            map_json(self.add_record, self.input_file_path)
        else:
            raise ValueError("Files must be either xml or json")

    def add_record(self, record):
        mr = MarcRecord(record)
        data = (
            mr.id(),
            mr.title() or None,
            mr.author() or None,
            mr.publication_year() or None,
            mr.pagination() or None,
            mr.edition() or None,
            mr.publisher_name() or None,
            mr.type_of() or None,
            mr.is_electronic_resource(),
            self.source_file,
        )
        try:
            self.cursor.execute(CREATE_RECORD_SQL, data)
        except psycopg2.DatabaseError:
            pass

1	import os.path	1✔
2	import time	1✔
3	import psycopg2	1✔
4	from pymarc import map_xml	1✔
5	from config import settings	1✔
6	from src.marc_record import MarcRecord	1✔
7	from src.streaming_json_handler import map_json	1✔
8
9	CREATE_TABLE_SQL = """CREATE TABLE IF NOT EXISTS records (	1✔
10	id TEXT,
11	title TEXT,
12	author TEXT,
13	publication_year TEXT,
14	pagination TEXT,
15	edition TEXT,
16	publisher_name TEXT,
17	type_of VARCHAR,
18	is_electronic_resource BOOL,
19	source_file TEXT,
20	UNIQUE (id)
21	);
22	"""
23
24	CREATE_RECORD_SQL = """INSERT INTO records VALUES	1✔
25	(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
26	"""
27
28
29	class MarcToDb:	1✔
30	conn = psycopg2.connect(	1✔
31	database=settings.db_name,
32	user=settings.db_user,
33	host=settings.db_host,
34	port=settings.db_port,
35	)
36	conn.autocommit = True	1✔
37
38	@classmethod	1✔
39	def find_or_create_table(cls):	1✔
40	with cls.conn.cursor() as cur:	1✔
41	cur.execute(CREATE_TABLE_SQL)	1✔
42
43	def __init__(self, input_file_path):	1✔
44	self.input_file_path = input_file_path	1✔
45	self.conn = MarcToDb.conn	1✔
46	self.source_file, self.file_extension = os.path.splitext(	1✔
47	os.path.basename(self.input_file_path)
48	)
49	self.cursor = self.conn.cursor()	1✔
50
51	def to_db(self):	1✔
52	print(	1✔
53	f"""time: {time.asctime(time.localtime())} -
54	writing records in {self.input_file_path} to database
55	"""
56	)
57	if self.file_extension == ".xml":	1✔
58	map_xml(self.add_record, self.input_file_path)	1✔
59	elif self.file_extension == ".json":	1✔
60	map_json(self.add_record, self.input_file_path)	1✔
61	else:
NEW 62	raise ValueError("Files must be either xml or json")	×
63
64	def add_record(self, record):	1✔
65	mr = MarcRecord(record)	1✔
66	data = (	1✔
67	mr.id(),
68	mr.title() or None,
69	mr.author() or None,
70	mr.publication_year() or None,
71	mr.pagination() or None,
72	mr.edition() or None,
73	mr.publisher_name() or None,
74	mr.type_of() or None,
75	mr.is_electronic_resource(),
76	self.source_file,
77	)
78	try:	1✔
79	self.cursor.execute(CREATE_RECORD_SQL, data)	1✔
80	except psycopg2.DatabaseError:	1✔
81	pass	1✔

pulibrary / pymarc_dedupe / c1019695-ebf0-419c-9312-e0b247a62c6a

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous