Skip to content

S
sqlite with fastapi
Commit a5e6bd1b authored by ramya.r's avatar ramya.r
Browse files
Options

first

parent c263356d
Show whitespace changes
Inline Side-by-side
Showing with 181 additions and 139 deletions
.idea/sonarlint/issuestore/index.pb
View file @ a5e6bd1b
^
..idea/inspectionProfiles/profiles_settings.xml,1\e\1e9075f5bf079c01ef2c910709e91a497d262080
\
,.idea/inspectionProfiles/Project_Default.xml,4\9\496a238a6afa168dbaf6efd37bb459331589579c
w
G.idea/sonarlint/issuestore/4/0/40143b61ec022045fa562c6d31ea2c886dd2f4b5,e\4\e4e6c66b897df14d56a75c299879dbe09d5dc0f8

Q.idea/sonarlint/securityhotspotstore/4/0/40143b61ec022045fa562c6d31ea2c886dd2f4b5,7\5\75fce93979f8f7d0b463a17996643c419b5503f5
E
employee_details.json,2\6\26d582e31ae3bfa82a16a35d19aa5678e0598026
4
.env,3\c\3c84dcdc6bbe3d7817c49dcdc327b926fea1808a
\ No newline at end of file
.idea/sonarlint/securityhotspotstore/index.pb
View file @ a5e6bd1b
^
..idea/inspectionProfiles/profiles_settings.xml,1\e\1e9075f5bf079c01ef2c910709e91a497d262080
\
,.idea/inspectionProfiles/Project_Default.xml,4\9\496a238a6afa168dbaf6efd37bb459331589579c
w
G.idea/sonarlint/issuestore/4/0/40143b61ec022045fa562c6d31ea2c886dd2f4b5,e\4\e4e6c66b897df14d56a75c299879dbe09d5dc0f8

Q.idea/sonarlint/securityhotspotstore/4/0/40143b61ec022045fa562c6d31ea2c886dd2f4b5,7\5\75fce93979f8f7d0b463a17996643c419b5503f5
E
employee_details.json,2\6\26d582e31ae3bfa82a16a35d19aa5678e0598026
4
.env,3\c\3c84dcdc6bbe3d7817c49dcdc327b926fea1808a
\ No newline at end of file
.idea/vcs.xml 0 → 100644
View file @ a5e6bd1b
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="$PROJECT_DIR$" vcs="Git" />
</component>
</project>
\ No newline at end of file
internsb2.db
View file @ a5e6bd1b
No preview for this file type
main.py
View file @ a5e6bd1b
import uvicorn
from fastapi import FastAPI
from dotenv import load_dotenv
from script.service.userdetails import router
from script.service.user_details import router
load_dotenv()
app = FastAPI(title="Employees")
app.include_router(router)
if __name__ == "__main__":
uvicorn.run("main:app", host='localhost', port=9000)
uvicorn.run("main:app", host='localhost', port=8000)
script/config/app_config.py
View file @ a5e6bd1b
import os
from configparser import SafeConfigParser
import sqlite3
from configparser import ConfigParser
config = SafeConfigParser()
config = ConfigParser()
config.read('conf/application.conf')
class Service:
port = int(config.getint("service", "port"))
host = config.get("service", "host")
port = config.getint('service', 'port')
host = config.get('service', 'host')
class Sqlite:
db_uri = os.environ.get('DB_URI')
db_uri= os.environ.get("DB_URI")
def create_sqlite_connection():
conn = sqlite3.connect("internsb2.db", check_same_thread=False)
return conn
script/constants/__init__.py
View file @ a5e6bd1b
......@@ -2,6 +2,11 @@ class APIEndpoints:
save = "/save"
employee_base = "/employees"
insert = "/insert"
subordinates= "/subordinates"
subordinates = "/subordinates"
find = "/find"
class Place:
cities = ['New York', 'Los Angeles', 'Chicago', 'Houston', 'Phoenix', 'Philadelphia', 'San Antonio',
'San Diego', 'Dallas', 'San Jose', 'Austin', 'Jacksonville', 'Fort Worth', 'Columbus',
'San Francisco', 'Charlotte', 'Indianapolis', 'Seattle', 'Denver', 'Washington']
script/core/handlers/emp_info.py 0 → 100644
View file @ a5e6bd1b
import numpy as np
import pandas as pd
from sqlalchemy import MetaData
from sqlalchemy.testing.schema import Table
from script.constants import Place
from script.utils.sqlite_connection import engine
from script.utils.sqlite_utils import Utils
my_data = Utils
class Userdetails:
@staticmethod
def create_user(data):
try:
employee_df = pd.DataFrame(data["employees"])
employee_df["city"] = np.random.choice(Place.cities, size=len(employee_df))
for _, employee in employee_df.iterrows():
Utils.create_info(id=employee["id"], name=employee["name"], city=employee["city"])
subordinates = employee_df.loc[employee_df["id"].isin(employee["subordinates"]), "name"].tolist()
subordinate_name = ", ".join(subordinates)
id = employee['id']
user_id = employee['id']
Utils.insert_into_subordinate(id, user_id, employee["subordinates"], subordinate_name)
return {"message": "User info created successfully"}
except Exception as e:
print(e, "Error Detected in inserting")
@staticmethod
def find_data():
try:
conn = engine.connect()
metadata = MetaData()
user_info = Table('user_info', metadata, autoload=True, autoload_with=engine)
subordinates_info = Table('subordinates_info', metadata, autoload=True, autoload_with=engine)
user_info_columns = [col.name for col in user_info.columns]
subordinate_columns = [col.name for col in subordinates_info.columns]
user_info_query = user_info.select()
user_info_rows = conn.execute(user_info_query).fetchall()
user_info_body = [{user_info_columns[i]: row[i] for i in range(len(user_info_columns))} for row in
user_info_rows]
subordinate_query = subordinates_info.select()
subordinate_rows = conn.execute(subordinate_query).fetchall()
subordinate_body = [{subordinate_columns[i]: row[i] for i in range(len(subordinate_columns))} for row in
subordinate_rows]
conn.close()
return {"header": user_info_columns, "body": user_info_body, "subordinate_header": subordinate_columns,
"subordinate_body": subordinate_body}
except Exception as e:
print(e, "Error detected in finding employee data")
\ No newline at end of file
script/core/handlers/sqlite.py deleted 100644 → 0
View file @ c263356d
import json
import random
import sqlite3
from script.utils.sqlite_utils import Utils
from script.utils.sqlite_connection import create_sqlite_connection
my_data = Utils
with open("employee_details.json") as f:
data = json.load(f)
class Userdetails:
@staticmethod
def create_user():
try:
cities = ['New York', 'Los Angeles', 'Chicago', 'Houston', 'Phoenix', 'Philadelphia', 'San Antonio',
'San Diego', 'Dallas', 'San Jose', 'Austin', 'Jacksonville', 'Fort Worth', 'Columbus',
'San Francisco', 'Charlotte', 'Indianapolis', 'Seattle', 'Denver', 'Washington']
for employee in data["employees"]:
employee["city"] = random.choice(cities)
Utils.create_info(id=employee["id"], name=employee["name"], city=employee["city"])
subordinates = []
for subordinate_id in employee["subordinates"]:
subordinate = next(
(subordinate for subordinate in data["employees"] if subordinate["id"] == subordinate_id), None)
if subordinate is not None:
subordinates.append(subordinate["name"])
subordinate_name = ", ".join(subordinates)
id = employee['id']
user_id = employee['id']
Utils.insert_into_subordinate(id, user_id, employee["subordinates"], subordinate_name)
return {"message": "User info created successfully"}
except Exception as e:
print(e, "Error Detected in inserting")
@staticmethod
def find_data():
try:
conn = create_sqlite_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM user_info")
columns = cursor.description
header = [{"label": col[0].replace("_", " ").title(), "value": col[0]} for col in columns]
cursor.execute("SELECT * FROM user_info")
rows = cursor.fetchall()
user_info = [{"id": row[0], "name": row[1], "city": row[2]} for row in rows]
cursor.execute("SELECT * FROM subordinates_info")
columns = cursor.description
subordinate_header = [{"label": col[0].replace("_", " ").title(), "value": col[0]} for col in columns]
cursor.execute("SELECT * FROM subordinates_info")
rows = cursor.fetchall()
subordinate_info = [{"id": row[0], "user_id": row[1], "subordinate_ids": row[2], "subordinate_name": row[3]}
for row in rows]
conn.close()
return {"header": header, "body": user_info, "subordinate_header": subordinate_header,
"subordinate_body": subordinate_info}
except Exception as e:
print(e, "Error detected in finding employee data")
script/core/schema/model.py
View file @ a5e6bd1b
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from script.utils.sqlite_connection import create_sqlite_connection
class Employee:
@staticmethod
def create_tables():
conn = create_sqlite_connection()
conn.execute("""
CREATE TABLE IF NOT EXISTS user_info (
id INTEGER PRIMARY KEY,
name TEXT,
city TEXT
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS subordinates_info (
id INTEGER PRIMARY KEY AUTOINCREMENT ,
user_id INTEGER,
subordinate_ids STRING NULL,
subordinate_name TEXT NOT NULL,
FOREIGN KEY(user_id) REFERENCES user_info(id)
)
""")
Base = declarative_base()
class User(Base):
__tablename__ = 'user_info'
id = Column(Integer, primary_key=True)
name = Column(String)
city = Column(String)
class Subordinates(Base):
__tablename__ = 'subordinates_info'
id = Column(Integer, primary_key=True, autoincrement=True)
user_id = Column(Integer, ForeignKey('user_info.id'))
subordinate_ids = Column(String, nullable=True)
subordinate_name = Column(String, nullable=False)
script/service/userdetails.py → script/service/user_details.py
View file @ a5e6bd1b
import json
import logging
from fastapi import APIRouter
from fastapi import APIRouter, UploadFile, File
from script.constants import APIEndpoints
from script.core.handlers.sqlite import Userdetails
from script.core.handlers.emp_info import Userdetails
from script.core.schema.response import DefaultResponse
from script.utils.sqlite_utils import Utils
user_details = Userdetails()
router = APIRouter(prefix=APIEndpoints.employee_base)
handler = Userdetails()
data = {}
@router.post("/upload")
async def upload_file(file: UploadFile = File()):
global data
try:
contents = file.file.read()
data = json.loads(contents)
except Exception as e:
print(e, "uploading failed")
@router.post("/create_db")
async def db_creation():
try:
Utils.create_db()
return "db created"
except Exception as e:
print(e, "db not created")
@router.post(APIEndpoints.insert)
def insert_item():
async def insert_item():
try:
handler.create_user()
handler.create_user(data)
return DefaultResponse(message="Successfully Inserted", status="success")
except ValueError:
return DefaultResponse(message="Due to value error")
......@@ -23,6 +44,7 @@ def insert_item():
logging.exception(e)
return DefaultResponse(message="Insertion Failed due to server error")
@router.get(APIEndpoints.find)
async def find_item():
try:
......@@ -33,5 +55,3 @@ async def find_item():
except Exception as e:
logging.exception(e)
return DefaultResponse(message="Finding Failed due to server error")
script/utils/sqlite_connection.py
View file @ a5e6bd1b
import sqlite3
from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import sessionmaker
from script.config.app_config import create_sqlite_connection
db = create_sqlite_connection()
engine = create_engine('sqlite:///internsb2.db',echo =True)
LocalSession = sessionmaker(bind=engine)
session = LocalSession()
metadata = MetaData()
def create_sqlite_connection():
conn = sqlite3.connect("internsb2.db",check_same_thread=False)
return conn
script/utils/sqlite_utils.py
View file @ a5e6bd1b
import sqlite3
from script.core.schema.model import Employee
from script.utils.sqlite_connection import create_sqlite_connection
conn = create_sqlite_connection()
cur = conn.cursor()
Employee.create_tables()
from script.core.schema.model import User, Subordinates, Base
from script.utils.sqlite_connection import session,engine
class Utils:
@staticmethod
def create_info(id: int, name: str, city: str):
conn = sqlite3.connect("internsb2.db")
cursor = conn.cursor()
cursor.execute("INSERT INTO user_info (id, name, city) VALUES (?, ?,?)", (id, name, city))
conn.commit()
new_employee = User(id=id, name=name, city=city)
session.add(new_employee)
session.commit()
return {"message": "User info created successfully"}
@staticmethod
def insert_into_subordinate(id, user_id, subordinate_ids, subordinate_name):
try:
create_sqlite_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO subordinates_info (id, user_id, subordinate_name, subordinate_ids) VALUES (?, ?, ?, ?)",
(id, user_id, subordinate_name, ",".join(str(i) for i in subordinate_ids)))
conn.commit()
conn.close()
print("Subordinate IDs:", ", ".join(str(i) for i in subordinate_ids))
subordinates = Subordinates(id=id, user_id=user_id, subordinate_name=subordinate_name,
subordinate_ids=",".join(str(i) for i in subordinate_ids))
session.add(subordinates)
session.commit()
except Exception as e:
print(e)
@staticmethod
def find_query(user_id):
cursor = conn.cursor()
cursor.execute("SELECT id, name, city FROM Employee WHERE id=?", (user_id,))
user_data = cursor.fetchall()
cursor.execute("SELECT id, user_id, subordinate_ids, subordinate_name FROM Subordinates WHERE user_id=?",
(user_id,))
subordinates_data = cursor.fetchall()
user_data = session.query(User).filter(User.id == user_id).all()
subordinates_data = session.query(Subordinates).filter(Subordinates.user_id == user_id).all()
list_ = []
for data in user_data:
list_.append({"id": data[0], "name": data[1], "city": data[2]})
list_.append({"id": data.id, "name": data.name, "city": data.city})
for data in subordinates_data:
list_.append({"id": data[0], "user_id": data[1], "subordinate_ids": data[2], "subordinate_name": data[3]})
conn.close()
list_.append({"id": data.id, "user_id": data.user_id, "subordinate_ids": data.subordinate_ids,
"subordinate_name": data.subordinate_name})
session.close()
return list_
return {"message": "successfully founded"}
@staticmethod
def create_db():
Base.metadata.create_all(engine)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment