Merge pull request #18 from KAnanev/dev

This pull request closes #16

Makefile

@@ -31,4 +31,7 @@ start-testdb:
 stop-testdb:
  docker-compose -f docker-compose.test.yml down
 
-.PHONY: install test lint selfcheck check build dev start start-testdb stop-testdb
+init_db:
+ poetry run flask --app page_analyzer:app init-db
+
+.PHONY: install test lint selfcheck check build dev start start-testdb stop-testdb init_db

page_analyzer/database.sql

@@ -1,7 +1,19 @@
 DROP TABLE IF EXISTS urls CASCADE;
+DROP TABLE IF EXISTS url_checks CASCADE;
 
 CREATE TABLE IF NOT EXISTS urls (
  id SERIAL PRIMARY KEY,
  name TEXT UNIQUE NOT NULL,
  created_at TIMESTAMP WITH TIME ZONE
+);
+
+CREATE TABLE IF NOT EXISTS url_checks (
+ id SERIAL PRIMARY KEY,
+ url_id INT,
+ status_code TEXT,
+ h1 TEXT,
+ title TEXT,
+ description TEXT,
+ created_at TIMESTAMP WITH TIME ZONE,
+ FOREIGN KEY (url_id) REFERENCES urls (id)
 );

page_analyzer/db.py

@@ -1,17 +1,15 @@
-import psycopg
 import click
 
 from flask import current_app, g
-from psycopg.rows import dict_row
+
+from page_analyzer.services.db import PostgresDB
 
 
 def get_db():
  """Функция проверяет наличие объекта соединения
  с базой данных в контексте приложения Flask"""
  if 'db' not in g:
- g.db = psycopg.connect(
- current_app.config['DATABASE_URL'], row_factory=dict_row
- )
+ g.db = PostgresDB(current_app.config['DATABASE_URL'])
  return g.db
 
 
@@ -26,8 +24,7 @@ def init_db():
  db = get_db()
 
  with current_app.open_resource('database.sql') as f:
- db.execute(f.read().decode('utf8'))
- db.commit()
+ db.execute_query(f.read().decode('utf8'))
 
 
 @click.command('init-db')

page_analyzer/models.py

@@ -0,0 +1,46 @@
+from datetime import datetime
+from typing import Optional, List, Dict, Union
+from urllib.parse import urlparse
+
+import validators
+from pydantic import BaseModel, field_validator, Field
+
+from page_analyzer.services.utils import get_date_now
+
+URL_MAX_LENGTH = 255
+
+
+class URLBaseMixin(BaseModel):
+ id: Optional[int] = None
+ created_at: datetime = Field(default_factory=get_date_now)
+
+
+class URLModel(URLBaseMixin):
+ name: str = Field(max_length=URL_MAX_LENGTH)
+
+ @field_validator('name')
+ def validate_url(cls, value: str) -> str:
+
+ if not validators.url(value):
+ raise ValueError('Invalid url address')
+ url = urlparse(value)
+ return f'{url.scheme}:https://{url.netloc}'.lower()
+
+
+class URLChecks(URLBaseMixin):
+ url_id: int
+ status_code: Optional[int] = None
+ h1: Optional[str] = None
+ title: Optional[str] = None
+ description: Optional[str] = None
+
+
+class URLSModel(URLModel):
+ url_checks: Optional[List[URLChecks]] = []
+
+ @field_validator('url_checks')
+ def validate_url_checks(cls, value: List) -> List[
+ Dict[str, Union[str, int]]
+ ]:
+ value.sort(key=lambda x: -x.id)
+ return value

page_analyzer/services/check_url.py

@@ -0,0 +1,33 @@
+import logging
+
+from page_analyzer.models import URLChecks
+from page_analyzer.services.db import PostgresDB
+
+INSERT_CHECK_URL = """INSERT INTO
+url_checks (url_id, status_code, h1, title, description, created_at)
+VALUES (%s, %s, %s, %s, %s, %s);"""
+
+
+class CheckURLService:
+
+ def __init__(self, db: PostgresDB):
+ self.db = db
+ self.logger = logging.getLogger(__name__)
+
+ def check(self, url_id):
+ result = self.check_url(url_id)
+ if result:
+ self.db.execute_query(INSERT_CHECK_URL, (
+ result.url_id,
+ result.status_code,
+ result.h1,
+ result.title,
+ result.description,
+ result.created_at,
+ ), commit=True)
+ return 'Страница успешно проверена', 'success'
+ return 'Произошла ошибка при проверке', 'danger'
+
+ @staticmethod
+ def check_url(url_id):
+ return URLChecks(url_id=url_id)

page_analyzer/services/db.py

@@ -0,0 +1,56 @@
+import logging
+from typing import List, Tuple, Any, Union, Dict
+
+import psycopg
+from psycopg.rows import dict_row
+
+
+class PostgresDB:
+ def __init__(self, dsn):
+
+ self.logger = logging.getLogger(__name__)
+
+ try:
+ self.connection = psycopg.connect(dsn, row_factory=dict_row)
+ except Exception as e:
+ self.logger.error(f"Ошибка при подключении к базе данных: {str(e)}")
+
+ def close(self) -> None:
+
+ """Закрывает соединение с базой данных"""
+
+ self.connection.close()
+ self.logger.info("Соединение с базой данных закрыто")
+
+ def cursor(self):
+ """Запрос к бд"""
+
+ return self.connection.cursor()
+
+ def _execute(self, query, params):
+ return self.cursor().execute(query, params)
+
+ def execute_query(
+ self,
+ query: str,
+ params: Tuple[Any, ...] = None,
+ many=False,
+ commit=False
+ ) -> Union[List[Dict[str, Any]], Dict[str, Any], None]:
+
+ result = self._execute(query, params)
+
+ if result.description:
+ if many:
+ result = result.fetchall()
+ else:
+ result = result.fetchone()
+
+ if commit:
+ self.connection.commit()
+
+ return result
+
+ def is_closed(self):
+ if self.connection:
+ return self.connection.closed

page_analyzer/services/url.py

@@ -0,0 +1,101 @@
+from typing import List, Optional
+
+from page_analyzer.models import URLModel, URLSModel
+from page_analyzer.services.db import PostgresDB
+
+GET_ITEMS = """SELECT
+ json_build_object(
+ 'id', id,
+ 'name', name,
+ 'created_at', created_at
+) AS result
+FROM urls;"""
+
+GET_JSON_BY_ID = """SELECT
+ json_build_object(
+ 'id', urls.id,
+ 'name', urls.name,
+ 'created_at', urls.created_at,
+ 'url_checks', COALESCE(json_agg(json_build_object(
+ 'id', url_checks.id,
+ 'url_id', url_checks.url_id,
+ 'status_code', url_checks.status_code,
+ 'h1', url_checks.h1,
+ 'title', url_checks.title,
+ 'description', url_checks.description,
+ 'created_at', url_checks.created_at
+ )) FILTER (WHERE url_checks.id IS NOT NULL) , '[]'::json)
+ ) AS result
+FROM urls
+LEFT JOIN url_checks ON urls.id = url_checks.url_id
+WHERE urls.id = (%s)
+GROUP BY urls.id;"""
+
+GET_JSON_BY_URL = """SELECT
+ json_build_object(
+ 'id', id,
+ 'name', name,
+ 'created_at', created_at
+) AS result
+FROM urls WHERE name = (%s)"""
+
+INSERT_ITEM_RETURN_JSON = """INSERT INTO
+ urls (name, created_at)
+ VALUES (%s,%s)
+ RETURNING json_build_object(
+ 'id', id,
+ 'name', name,
+ 'created', created_at
+ ) AS result;"""
+
+
+class URLService:
+ def __init__(self, db: PostgresDB):
+ self.db = db
+
+ def get_json_by_id(self, url_id: int) -> Optional[URLSModel]:
+ item = self.db.execute_query(GET_JSON_BY_ID, (url_id,), )
+ if not item:
+ return None
+ return URLSModel(**item['result'])
+
+ def get_all_urls(self) -> List[URLModel] | None:
+ items = self.db.execute_query(GET_ITEMS, many=True)
+ if items:
+ sorted_items = sorted(items, key=lambda item: -item['result']['id'])
+ items = [URLModel(**item['result']) for item in sorted_items]
+ return items
+
+ def _get_url_id_by_url_name(self, item: URLModel) -> Optional[URLModel]:
+
+ exist_item = self.db.execute_query(GET_JSON_BY_URL, (item.name,), )
+ if exist_item:
+ item = URLModel(**exist_item['result'])
+ return item
+
+ def insert_url(self, url: str) -> dict[str, tuple[str, str] | URLModel]:
+
+ item = None
+
+ try:
+ item = URLModel(name=url)
+ item = self._get_url_id_by_url_name(item)
+
+ if item.id:
+ message = ('Страница уже существует', 'info')
+
+ else:
+ raw_item = self.db.execute_query(
+ INSERT_ITEM_RETURN_JSON,
+ (item.name, item.created_at),
+ commit=True
+ )
+ item = URLModel(**raw_item['result'])
+ message = ('Страница успешно добавлена', 'success')
+ except ValueError:
+ message = ('Некорректный URL', 'danger')
+
+ return {
+ 'item': item,
+ 'message': message,
+ }

page_analyzer/services/utils.py

@@ -0,0 +1,6 @@
+from datetime import datetime
+
+
+def get_date_now():
+ now = datetime.now()
+ return now.strftime('%Y-%m-%d %H:%M:%S')