Commit 0772a750 authored by Luciano Cauzzi's avatar Luciano Cauzzi

Initial commit

parents
Pipeline #60 canceled with stages
# Created by .ignore support plugin (hsz.mobi)
### Python template
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
*.egg-info/
.installed.cfg
*.egg
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*,cover
.hypothesis/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# IPython Notebook
.ipynb_checkpoints
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# dotenv
.env
# virtualenv
venv/
ENV/
# Spyder project settings
.spyderproject
# Rope project settings
.ropeproject
### VirtualEnv template
# Virtualenv
# http://iamzed.com/2009/05/07/a-primer-on-virtualenv/
.Python
[Bb]in
[Ii]nclude
[Ll]ib
[Ll]ib64
[Ll]ocal
[Ss]cripts
pyvenv.cfg
.venv
pip-selfcheck.json
### JetBrains template
# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio and Webstorm
# Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839
# User-specific stuff:
.idea/workspace.xml
.idea/tasks.xml
.idea/dictionaries
.idea/vcs.xml
.idea/jsLibraryMappings.xml
# Sensitive or high-churn files:
.idea/dataSources.ids
.idea/dataSources.xml
.idea/dataSources.local.xml
.idea/sqlDataSources.xml
.idea/dynamic.xml
.idea/uiDesigner.xml
# Gradle:
.idea/gradle.xml
.idea/libraries
# Mongo Explorer plugin:
.idea/mongoSettings.xml
.idea/
## File-based project format:
*.iws
## Plugin-specific files:
# IntelliJ
/out/
# mpeltonen/sbt-idea plugin
.idea_modules/
# JIRA plugin
atlassian-ide-plugin.xml
# Crashlytics plugin (for Android Studio and IntelliJ)
com_crashlytics_export_strings.xml
crashlytics.properties
crashlytics-build.properties
fabric.properties
from .container import NetworkManager
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
from logger_library import Logger
logger = Logger.app_logger().get_logger()
class AppNetworkManager:
def __init__(self, total=10, read=10, connect=7, backoff_factor=0.5):
retry = Retry(
total=total,
read=read,
connect=connect,
backoff_factor=backoff_factor
)
self.__session = requests.Session()
adapter = HTTPAdapter(max_retries=retry)
self.__session.mount('http://', adapter)
self.__session.mount('https://', adapter)
pass
from dependency_injector import providers, containers
from network_manager.app_network_manager import AppNetworkManager
from network_manager.pyramid_manager import AppPyramidManager
class NetworkManager(containers.DeclarativeContainer):
manager = providers.Singleton(AppNetworkManager, total=10, read=10, connect=7, backoff_factor=0.5)
class PyramidManager(containers.DeclarativeContainer):
config = providers.Configuration('config')
manager = providers.Factory(AppPyramidManager, base_url=config.pyramid.base_url, resource_url=config.pyramid.res_url, token=config.pyramid.token)
from logger_library import Logger
from .pyramid_model import *
logger = Logger.app_logger().get_logger()
class AppPyramidManager:
def __init__(self, base_url, resource_url, token):
self.__url__ = resource_url
self.__token__ = token
self.__base_url__ = base_url
logger.debug("Initialization completed: \nBASE_URL= {} \nRESOURCE_URL= {}\nTOKEN= {}".format(base_url, resource_url, token))
from dataclasses import dataclass
from typing import Any, Optional, TypeVar, cast, Type, Callable, List
T = TypeVar("T")
def from_str(x: Any) -> str:
assert isinstance(x, str)
return x
def from_bytes(x: Any) -> bytes:
assert isinstance(x, bytes)
return x
def from_int(x: Any) -> int:
assert isinstance(x, int) and not isinstance(x, bool)
return x
def to_class(c: Type[T], x: Any) -> dict:
assert isinstance(x, c)
return cast(Any, x).to_dict()
def from_list(f: Callable[[Any], T], x: Any) -> List[T]:
assert isinstance(x, list)
return [f(y) for y in x]
@dataclass
class AppToken:
expire: int
token: str
@staticmethod
def from_dict(obj: Any) -> 'AppToken':
assert isinstance(obj, dict)
expire = from_int(obj.get("expire"))
token = from_str(obj.get("token"))
return AppToken(expire, token)
def to_dict(self) -> dict:
result: dict = {"expire": from_int(self.expire), "token": from_str(self.token)}
return result
from network_manager.container import PyramidManager
RES_URL = "v1/es/tributi/ingiunzioni"
BASE_URL = "http://pyramid-test.comune.brescia.it:8091/api"
TOKEN = "Xb4hQ1JT5DrMiYM91JoqwjMsU4xXuZVdb3q6MvFfVzmO70XR1xYYLIH4hv3qIGNL8VB1imOGYOQs6tBJKIY41QHsUCOa9MGOXxDMOG3T0Djm1r0a18eeZwJbSDJxnKCK"
def main():
print("Hello network")
# , base_url=config.pyramid.base_url, resource_url=config.pyramid.res_url, token=config.pyramid.token
config = {
'pyramid': {
'base_url': BASE_URL,
'res_url': RES_URL,
'token': TOKEN,
}
}
manager = PyramidManager(config=config).manager()
pass
if __name__ == "__main__":
main()
from setuptools import setup
import setuptools
setup(
name='network_manager',
version='0.1',
packages=setuptools.find_packages(),
dependency_links=['http://github.comune.brescia.it/innovation-that-excites/logger_library#egg=logger_library'],
install_requires=['logger_library', 'dependency_injector', 'requests'],
url='http://github/innovation-that-excites/network_manager',
license='MIT',
author='Luciano Cauzzi',
author_email='lcauzzi@comune.brescia.it',
description=''
)
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