Add db.py to handle slices in database

swagger/__pycache__/

src/__pycache__/

venv/

.env

from src.config.constants import NSC_PORT

from src.webui.gui import gui_bp

from src.config.config import create_config

from src.database.db import init_db

def create_app():

    """Factory para crear la app Flask con la configuración cargada"""

    init_db()

    app = Flask(__name__)

    app = create_config(app)

    CORS(app)

# This file is an original contribution from Telefonica Innovación Digital S.L.

from src.config.constants import DATABASE_PATH

from src.utils.send_response import send_response

import os, json, logging

import logging

from flask import current_app

from src.database.db import get_data, delete_data, get_all_data, delete_all_data

class Api:

    def __init__(self, slice_service):

        """

        try:

            # Read slice database from JSON file

            with open(os.path.join(DATABASE_PATH, "slice_ddbb.json"), 'r') as file:

                content = json.load(file)

            content = get_all_data()

            # If specific slice ID is provided, find and return matching slice

            if slice_id:

                for slice in content:

            - If need_l2vpn_support is True, performs additional L2VPN cleanup

        """

        try:

            # Read current slice database

            with open(os.path.join(DATABASE_PATH, "slice_ddbb.json"), 'r') as file:

                content = json.load(file)

            id = None

            # Delete specific slice if slice_id is provided

            if slice_id:

                for i, slice in enumerate(content):

                    if slice["slice_id"] == slice_id and slice.get("controller") == self.slice_service.controller_type:

                        del content[i]

                        id = i

                        break

                slice = get_data(slice_id)

                # Raise error if slice not found

                if id is None:

                if not slice or slice.get("controller") != self.slice_service.controller_type:

                    raise ValueError("Transport network slice not found")

                # Update slice database

                with open(os.path.join(DATABASE_PATH, "slice_ddbb.json"), 'w') as file:

                    json.dump(content, file, indent=4)

                delete_data(slice_id)

                logging.info(f"Slice {slice_id} removed successfully")

                return {}, 204

                    if current_app.config["TFS_L2VPN_SUPPORT"]:

                        self.slice_service.tfs_l2vpn_delete()

                data_removed = [slice for slice in content if slice.get("controller") == self.slice_service.controller_type] 

                # Verify slices exist before deletion

                if len(data_removed) == 0:

                    raise ValueError("Transport network slices not found")

                filtered_data = [slice for slice in content if slice.get("controller") != self.slice_service.controller_type]    

                # Clear slice database

                with open(os.path.join(DATABASE_PATH, "slice_ddbb.json"), 'w') as file:

                    json.dump(filtered_data, file, indent=4)

                delete_all_data()

                logging.info("All slices removed successfully")

                return {}, 204

# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#     http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import sqlite3, json, logging

# Database file

DB_NAME = "slice.db"

# Initialize database and create table

def init_db():

    conn = sqlite3.connect(DB_NAME)

    cursor = conn.cursor()

    cursor.execute("""

        CREATE TABLE IF NOT EXISTS slice (

            slice_id TEXT PRIMARY KEY,

            intent TEXT NOT NULL,

            controller TEXT NOT NULL

        )

    """)

    conn.commit()

    conn.close()

# Save data to the database

def save_data(slice_id: str, intent_dict: dict, controller: str):

    intent_str = json.dumps(intent_dict)

    conn = sqlite3.connect(DB_NAME)

    cursor = conn.cursor()

    try:

        cursor.execute("INSERT INTO slice (slice_id, intent, controller) VALUES (?, ?, ?)", (slice_id, intent_str, controller))

        conn.commit()

    except sqlite3.IntegrityError:

        raise ValueError(f"Slice with id '{slice_id}' already exists.")

    finally:

        conn.close()

# Update data in the database

def update_data(slice_id: str, new_intent_dict: dict, controller: str):

    intent_str = json.dumps(new_intent_dict)

    conn = sqlite3.connect(DB_NAME)

    cursor = conn.cursor()

    cursor.execute("UPDATE slice SET intent = ?, controller = ? WHERE slice_id = ?", (intent_str, controller, slice_id))

    if cursor.rowcount == 0:

        raise ValueError(f"No slice found with id '{slice_id}' to update.")

    else:

        logging.debug(f"Slice '{slice_id}' updated.")

    conn.commit()

    conn.close()

# Delete data from the database

def delete_data(slice_id: str):

    conn = sqlite3.connect(DB_NAME)

    cursor = conn.cursor()

    cursor.execute("DELETE FROM slice WHERE slice_id = ?", (slice_id,))

    if cursor.rowcount == 0:

        raise ValueError(f"No slice found with id '{slice_id}' to delete.")

    else:

        logging.debug(f"Slice '{slice_id}' deleted.")

    conn.commit()

    conn.close()

# Get data from the database

def get_data(slice_id: str) -> dict[str, dict, str]:

    conn = sqlite3.connect(DB_NAME)

    cursor = conn.cursor()

    cursor.execute("SELECT * FROM slice WHERE slice_id = ?", (slice_id,))

    row = cursor.fetchone()

    conn.close()

    if row:

        column_names = [description[0] for description in cursor.description]

        result = dict(zip(column_names, row))

        if isinstance(result.get("intent"), str):

            try:

                result["intent"] = json.loads(result["intent"])

            except json.JSONDecodeError:

                raise Exception("Warning: 'intent' is not a valid JSON string.")

        return result

    else:

        raise ValueError(f"No slice found with id '{slice_id}'.")

# Get all slices

def get_all_data() -> dict[str, dict, str]:

    conn = sqlite3.connect(DB_NAME)

    cursor = conn.cursor()

    cursor.execute("SELECT * FROM slice")

    rows = cursor.fetchall()

    conn.close()

    return [

        {

        "slice_id": row[0],

        "intent": json.loads(row[1]),

        "controller": row[2] 

        }

    for row in rows

    ]

def delete_all_data():

    conn = sqlite3.connect(DB_NAME)

    cursor = conn.cursor()

    cursor.execute("DELETE FROM slice")

    conn.commit()

    conn.close()

    logging.debug("All slice data deleted.")

# Example usage

if __name__ == "__main__":

    init_db()

    # Save a slice

    test_intent = {"bandwidth": "1Gbps", "latency": "10ms", "provider": "opensec"}

    save_data("slice-001", test_intent, "TFS")

    # Get the slice

    result = get_data("slice-001")

    if result:

        print(f"Retrieved intent for slice-001: {result}")

    # Update the slice

    updated_intent = {"bandwidth": "2Gbps", "latency": "5ms", "provider": "opensec"}

    update_data("slice-001", updated_intent, "TFS")

    # Delete the slice

    delete_data("slice-001")

    get_all_data()

    delete_all_data()

[]

 No newline at end of file