data.datasets.low_flex_scenario.LowFlexScenario.__init__() - Code Metrics - Inspection of "Make sure that the connection settings are suitabl..." - openego/eGon-data - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#1345)

unknown

created 2025-08-11 09:51 UTC

LowFlexScenario.init() A

↳ Parent: data.datasets.low_flex_scenario

Complexity

Conditions

Size

Total Lines	15
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	14
dl	0
loc	15
rs	9.7
c	0
b	0
f	0
cc	1
nop	2

"""The central module to create low flex scenarios

"""
from airflow.models import Connection
from airflow.operators.postgres_operator import PostgresOperator
from airflow.settings import Session
from importlib_resources import files
from sqlalchemy.ext.declarative import declarative_base

from egon.data.datasets import Dataset

Base = declarative_base()


def ensure_postgres_connection():
    session = Session()

    conn = (
        session.query(Connection)
        .filter(Connection.conn_id == "egon_data")
        .one_or_none()
    )

    conn.conn_type = "postgres"

    session.commit()
    session.close()


class LowFlexScenario(Dataset):
    def __init__(self, dependencies):
        ensure_postgres_connection()
        super().__init__(
            name="low_flex_scenario",
            version="0.0.1",
            dependencies=dependencies,
            tasks=(
                {
                    PostgresOperator(
                        task_id="low_flex_eGon2035",
                        sql=files(__name__)
                        .joinpath("low_flex_eGon2035.sql")
                        .read_text(encoding="utf-8"),
                        postgres_conn_id="egon_data",
                        autocommit=True,
                    ),
                },
            ),
        )


1			"""The central module to create low flex scenarios
2
3			"""
4			from airflow.models import Connection
5			from airflow.operators.postgres_operator import PostgresOperator
6			from airflow.settings import Session
7			from importlib_resources import files
8			from sqlalchemy.ext.declarative import declarative_base
9
10			from egon.data.datasets import Dataset
11
12			Base = declarative_base()
13
14
15			def ensure_postgres_connection():
16			session = Session()
17
18			conn = (
19			session.query(Connection)
20			.filter(Connection.conn_id == "egon_data")
21			.one_or_none()
22			)
23
24			conn.conn_type = "postgres"
25
26			session.commit()
27			session.close()
28
29
30			class LowFlexScenario(Dataset):
31			def __init__(self, dependencies):
32			ensure_postgres_connection()
33			super().__init__(
34			name="low_flex_scenario",
35			version="0.0.1",
36			dependencies=dependencies,
37			tasks=(
38			{
39			PostgresOperator(
40			task_id="low_flex_eGon2035",
41			sql=files(__name__)
42			.joinpath("low_flex_eGon2035.sql")
43			.read_text(encoding="utf-8"),
44			postgres_conn_id="egon_data",
45			autocommit=True,
46			),
47			},
48			),
49			)
50

openego / eGon-data

Pull Request — dev (#1345)

LowFlexScenario.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

LowFlexScenario.init() A