Commit d6e5313b authored by Поляков Сергей Викторович's avatar Поляков Сергей Викторович
Browse files

3 Part.

Added endpoint POST for save to DB full link and return short link.
Web-service generated unique links.
Was make class BD_Manager for work with any DB from existed.
parent 2326f3e0
Pipeline #3656 failed with stages
in 3 minutes and 16 seconds
- full_link: https://habr.com/ru/company/ruvds/news/t/577872/
- full_link: https://www.kinopoisk.ru/
short_link: ex1
- full_link: https://habr.com/ru/company/ruvds/blog/576376/
- full_link: https://europaplus.ru/
short_link: li2
- full_link: https://habr.com/ru/company/ruvds/blog/577754/
- full_link: https://www.youtube.com/
short_link: en3
- full_link: https://habr.com/ru/company/ruvds/blog/576716/
- full_link: https://gitlab.mai.ru/PolyakovSV
short_link: zv4
- full_link: https://translate.yandex.ru/
short_link: tt1
from lib.yaml_lib import YamlManager
from lib.libs import Config
from pathlib import Path
import redis
conf = Config(Path("pref.json"))
yaml_manager = YamlManager(path=Path('data', conf.name_yaml_bd))
redis_cli = redis.Redis(host=conf.redis_host, port=conf.redis_port, db=conf.redis_db)
def export_data():
try:
for i in yaml_manager.read():
redis_cli.set(name=i['short_link'], value=i['full_link'])
print(f"Successfully added key: {i['short_link']} value: {i['full_link']}")
except Exception as e:
print(f'Error: {e}')
finally:
redis_cli.close()
export_data()
from lib.redis_lib import RedisManager
from lib.libs import LinkGenerator
class DB_Manager:
""" Main class for Read, Write, and Other methods with DB..."""
def __init__(self, config_class, yaml_manager_class):
self._conf = config_class
self._yaml_manager = yaml_manager_class
def key_exist(self, key):
"""Return True if key EXIST else False"""
if self._conf.storage == 'yaml':
if key in self._yaml_manager.get_keys():
return True
else:
return False
if self._conf.storage == 'redis':
with RedisManager(host=self._conf.redis_host, port=self._conf.redis_port, db=self._conf.redis_db) as t:
if t.key_exists(key):
return True
else:
return False
def get_by_key(self, key):
"""Return value by key"""
if self._conf.storage == 'yaml':
return self._yaml_manager.get_by_key(key)
if self._conf.storage == 'redis':
with RedisManager(host=self._conf.redis_host, port=self._conf.redis_port, db=self._conf.redis_db) as t:
return t.get_by_key(key)
def save(self, key: str, value: str):
if self._conf.storage == 'yaml':
self._yaml_manager.save(key=key, value=value)
if self._conf.storage == 'redis':
with RedisManager(host=self._conf.redis_host, port=self._conf.redis_port, db=self._conf.redis_db) as t:
t.save(key=key, value=value)
def get_keys(self):
if self._conf.storage == 'yaml':
return self._yaml_manager.get_keys()
if self._conf.storage == 'redis':
with RedisManager(host=self._conf.redis_host, port=self._conf.redis_port, db=self._conf.redis_db) as t:
return t.get_keys()
def unique_link(self, generate_method='uuid4'):
if generate_method =='uuid4':
method = 'uuid4'
if generate_method == 'timestamp':
method = 'timestamp'
while True:
link = LinkGenerator(generate_method=method).new_link
if not self.key_exist(link):
return link
if __name__ == "__main__":
from pathlib import Path
from lib.libs import Config, LinkGenerator
from lib.yaml_lib import YamlManager
from lib.redis_lib import RedisManager
conf = Config(Path("..", "pref.json"))
yaml_manager = YamlManager(path=Path("..", "data", conf.name_yaml_bd))
k = DB_Manager(conf, yaml_manager)
print(k.unique_link(generate_method='timestamp'))
from lib.redis_lib import RedisManager
from lib.yaml_lib import YamlManager
from lib.libs import Config
from pathlib import Path
conf = Config(Path("..", "pref.json"))
yaml_manager = YamlManager(path=Path("..", "data", conf.name_yaml_bd))
def export_data():
try:
for key in yaml_manager.get_keys():
value = yaml_manager.get_by_key(key)
print(f"Successfully added key: {key}, value: {value}")
with RedisManager(conf.redis_host, conf.redis_port, conf.redis_db) as f:
f.save(key,value)
except Exception as e:
print(f'Error: {e}')
if __name__ == "__main__":
export_data()
\ No newline at end of file
import datetime
import random
import uuid
import json
......@@ -24,25 +26,37 @@ class Config:
class LinkGenerator:
"""Takes the last generated link and returns a new in new_link variable"""
# сделал бы с помошью Unique но в ТЗ так
def __init__(self, last_link: str):
self._last_link = last_link
self._last_number = int(self._get_num()) + 1
self.new_link = (self._random_alpha() + str(self._last_number))
def _get_num(self):
"""Parses a string and returns only numbers"""
try:
t = ''
for i in self._last_link:
if i.isdigit():
t += i
"""Generating link and returns a new in new_link variable
generate_method = "timestamp" or "uuid4" (by default) return six characters
"""
def __init__(self, generate_method='uuid4'):
self._generate_method = generate_method
if self._generate_method == 'uuid4':
self.new_link = self._uuid4()
if self._generate_method == 'timestamp':
self.new_link = self._timestamp()
def _timestamp_now(self):
"""Return timestamp datetime now"""
return str(datetime.datetime.now().timestamp())
def _random_lower(self):
"""Return random letters lowercase"""
return chr(random.randint(97, 122))
def _random_upper(self):
"""Return random letters Uppercase"""
return chr(random.randint(65, 90))
def _timestamp(self):
ch = str(self._random_upper() + self._random_lower())
f = self._timestamp_now().replace('.', ch)[10:]
t = f[:3] + f[-3:]
return t
except Exception as e:
print(f'Error: {e}')
return ''
def _random_alpha(self):
"""Return two random letters"""
return chr(random.randint(97, 122)) + chr(random.randint(97, 122))
def _uuid4(self):
"""Returns first six characters"""
return str(uuid.uuid4())[:6]
......@@ -33,6 +33,12 @@ class RedisManager:
print(e)
return None
def get_keys(self):
_t = []
for i in self._redis_cli.keys(pattern='*'):
_t.append(i.decode('UTF-8'))
return _t
def __exit__(self, exc_type, exc_val, exc_tb):
""" Closing the connection """
self._redis_cli.close()
......
import os
import yaml
......@@ -7,32 +5,24 @@ class YamlManager:
def __init__(self, path):
"""path = path to db"""
self._path = path
if os.path.exists(self._path):
self.last_short_link = self.get_last_dict()['short_link']
else:
self.last_short_link = '0'
def save(self, short_link: str, full_link: str):
def save(self, key: str, value: str):
with open(self._path, 'a') as f:
yaml.dump([{"short_link": short_link, "full_link": full_link}], f, default_flow_style=False)
yaml.dump([{"short_link": key, "full_link": value}], f, default_flow_style=False)
def read(self):
"""Return all dict from yaml"""
try:
def get_keys(self):
with open(self._path) as f:
items = yaml.safe_load(f)
return items
except:
return None
dictionary_data = yaml.safe_load(f)
t = []
for i in dictionary_data:
t.append(i['short_link'])
return t
def get_by_key(self, key):
try:
for s in self.read():
if key == s['short_link']:
return s['full_link']
except:
return False
with open(self._path) as f:
dictionary_data = yaml.safe_load(f)
for i in dictionary_data:
if key == i['short_link']:
return i['full_link']
return None
def get_last_dict(self):
"""Return last element from yaml"""
return self.read()[-1]
from pathlib import Path
import uvicorn
from fastapi import FastAPI
from typing import Optional
from fastapi import FastAPI, Query
from starlette.responses import RedirectResponse
from lib.libs import Config
from lib.yaml_lib import YamlManager
from lib.redis_lib import RedisManager
from lib.db_manager import DB_Manager
conf = Config(Path("pref.json"))
yaml_manager = YamlManager(path=Path("data", conf.name_yaml_bd))
db = DB_Manager(config_class=conf, yaml_manager_class=yaml_manager)
app = FastAPI()
@app.get("/{shorted_link}")
async def short_link(shorted_link: str):
if conf.storage == 'yaml':
key = yaml_manager.get_by_key(shorted_link)
if key:
return RedirectResponse(url=key)
@app.get("/{link}")
async def short_link(link: str):
if db.key_exist(link):
return RedirectResponse(url=db.get_by_key(link))
else:
return {'404 Not found': link}
if conf.storage == 'redis':
with RedisManager(host=conf.redis_host, port=conf.redis_port, db=conf.redis_db) as t:
key = t.key_exists(shorted_link)
if key:
return RedirectResponse(url=t.get_by_key(shorted_link))
return '404 Not found'
@app.post("/generate_link/")
async def add_full_link(value: str, method: Optional[str] = Query(None)):
if method == 'timestamp':
key = db.unique_link(generate_method=method)
else:
key = db.unique_link()
db.save(key=key, value=value)
return {key: value}
if __name__ == "__main__":
......
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