Migrate database to sqlite

.gitignore

@@ -1,2 +1,2 @@
-src/constants.py
+constants.py
-data
+assets

app/__init__.py

@@ -0,0 +1,5 @@
+import logging
+
+logging.basicConfig(
+    format="[%(levelname) 5s/%(asctime)s] %(name)s: %(message)s", level=logging.WARNING
+)

app/bot.py

@@ -1,15 +1,9 @@
-import logging
-
 from telethon import TelegramClient, events
 from telethon.sessions import StringSession
 
-from constants import API_HASH, API_ID, BOT_TOKEN, SESSION
+from app.notifier import send_mail
-from database import search_database
+from constants import API_HASH, API_ID, BOT_TOKEN, EXAMPLE, SESSION
-from notifier import send_mail
+from database.crud import create_database, search_database
-
-logging.basicConfig(
-    format="[%(levelname) 5s/%(asctime)s] %(name)s: %(message)s", level=logging.WARNING
-)
 
 bot = TelegramClient(StringSession(SESSION), API_ID, API_HASH).start(
     bot_token=BOT_TOKEN
@@ -37,4 +31,5 @@ async def identify(event):
 
 
 if __name__ == "__main__":
+    create_database(data=EXAMPLE)
     bot.run_until_disconnected()

database/__init__.py

@@ -0,0 +1,9 @@
+from sqlalchemy import MetaData, create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+from constants import DATABASE
+
+engine = create_engine(DATABASE, connect_args={"check_same_thread": False})
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+Base = declarative_base()

database/crud.py

@@ -0,0 +1,27 @@
+from sqlalchemy import or_
+
+from database import SessionLocal, engine, models
+from database.models import Users
+
+db = SessionLocal()
+
+
+def search_database(id):
+    return db.query(Users).filter(
+        or_(Users.correo_institucional == id, Users.numero_de_documento == id)
+    )
+
+
+def insert_data(data):
+    item = Users(**data)
+    db.add(item)
+    db.commit()
+    db.refresh(item)
+
+
+def create_database(data):
+    models.Base.metadata.create_all(bind=engine)
+    existing_row = search_database(data["correo_institucional"])
+    if existing_row:
+        return
+    insert_data(data)

database/models.py

@@ -0,0 +1,17 @@
+from sqlalchemy import Column
+from sqlalchemy.types import Integer, String
+
+from database import Base
+
+
+class Users(Base):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    primer_apellido = Column(String, nullable=False)
+    segundo_apellido = Column(String, nullable=False)
+    nombre = Column(String, nullable=False)
+    tipo_de_documento = Column(String, nullable=False)
+    numero_de_documento = Column(String, nullable=False, unique=True)
+    centro_academico = Column(String, nullable=False)
+    correo_institucional = Column(String, nullable=False, unique=True)

shell.nix

@@ -6,9 +6,11 @@ mkShell {
   buildInputs = [
     python38
     python38Packages.telethon
+    python38Packages.sqlalchemy
     python38Packages.pytest
     python38Packages.isort
     python38Packages.pyflakes
+    sqlite
   ];
 
 }

src/database.py

@@ -1,15 +0,0 @@
-from csv import DictReader
-
-
-def find_user(reader, id):
-    for row in reader:
-        if row["numero_de_documento"] == id or row["correo_institucional"] == id:
-            return row
-    raise Exception("Error: User not found")
-
-
-def search_csv(filepath, id):
-    with open(filepath) as csvfile:
-        reader = DictReader(csvfile)
-        user = find_user(reader, id)
-        return user