14424792839

Committed 13 Apr 2025 01:04AM UTC coverage: 82.285%. First build

Build # 14424792839

Build Type

Pull #94

github

Committed by

web-flow

Commit Message

Merge 1d673cdce into 265e9c2b5

Pull Request Pull Request #94: Ingestion v2.0.0

Run Details

421 of 557 new or added lines in 12 files covered. (75.58%)

850 of 1033 relevant lines covered (82.28%)

0.82 hits per line

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

81.82

/src/database.py

import logging

import pandas as pd
from sqlalchemy.engine.base import Connection, Engine


def get_feature_flags(connection: Connection | Engine) -> pd.DataFrame:
    flags = pd.read_sql_query(sql="select * from marts.feature_flags;", con=connection)

    logging.info(f"Retrieving {len(flags)} Feature Flags")
    return flags


def check_feature_flag(flag: str, flags_df: pd.DataFrame) -> bool:
    flags_df = flags_df.query(f"flag == '{flag}'")

    if len(flags_df) > 0 and flags_df["is_enabled"].iloc[0] == 1:
        return True
    else:
        return False


def write_to_sql(con, table_name: str, df: pd.DataFrame, table_type: str) -> None:
    """
    Simple Wrapper Function to write a Pandas DataFrame to SQL

    Args:
        con (SQL Connection): The connection to the SQL DB.

        table_name (str): The Table name to write to SQL as.

        df (DataFrame): The Pandas DataFrame to store in SQL

        table_type (str): Whether the table should replace or append to an
            existing SQL Table under that name

    Returns:
        Writes the Pandas DataFrame to a Table in the Schema we connected to.

    """
    try:
        if len(df) == 0:
            logging.info(f"{table_name} is empty, not writing to SQL")
        else:
            df.to_sql(
                con=con,
                name=table_name,
                index=False,
                if_exists=table_type,
            )
            logging.info(
                f"Writing {len(df)} {table_name} rows to aws_{table_name}_source to SQL"
            )

        return None
    except Exception as error:
        logging.error(f"SQL Write Script Failed, {error}")
        return None
        # sentry_sdk.capture_exception(error)

1	import logging	1✔
2
3	import pandas as pd	1✔
4	from sqlalchemy.engine.base import Connection, Engine	1✔
5
6
7	def get_feature_flags(connection: Connection \| Engine) -> pd.DataFrame:	1✔
8	flags = pd.read_sql_query(sql="select * from marts.feature_flags;", con=connection)	1✔
9
10	logging.info(f"Retrieving {len(flags)} Feature Flags")	1✔
11	return flags	1✔
12
13
14	def check_feature_flag(flag: str, flags_df: pd.DataFrame) -> bool:	1✔
15	flags_df = flags_df.query(f"flag == '{flag}'")	1✔
16
17	if len(flags_df) > 0 and flags_df["is_enabled"].iloc[0] == 1:	1✔
18	return True	1✔
19	else:
20	return False	1✔
21
22
23	def write_to_sql(con, table_name: str, df: pd.DataFrame, table_type: str) -> None:	1✔
24	"""
25	Simple Wrapper Function to write a Pandas DataFrame to SQL
26
27	Args:
28	con (SQL Connection): The connection to the SQL DB.
29
30	table_name (str): The Table name to write to SQL as.
31
32	df (DataFrame): The Pandas DataFrame to store in SQL
33
34	table_type (str): Whether the table should replace or append to an
35	existing SQL Table under that name
36
37	Returns:
38	Writes the Pandas DataFrame to a Table in the Schema we connected to.
39
40	"""
41	try:	1✔
42	if len(df) == 0:	1✔
NEW 43	logging.info(f"{table_name} is empty, not writing to SQL")	×
44	else:
45	df.to_sql(	1✔
46	con=con,
47	name=table_name,
48	index=False,
49	if_exists=table_type,
50	)
51	logging.info(	1✔
52	f"Writing {len(df)} {table_name} rows to aws_{table_name}_source to SQL"
53	)
54
55	return None	1✔
NEW 56	except Exception as error:	×
NEW 57	logging.error(f"SQL Write Script Failed, {error}")	×
NEW 58	return None	×
59	# sentry_sdk.capture_exception(error)

jyablonski / python_docker / 14424792839

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