Why Core-Lib?
Core-Lib
was born to make the day-to-day work, The "work itself."
easy to master.
What is Core-Lib?
Core-Lib
is a framework for creating Python applications as libraries. It is essentially a POPO
(Plain Old Python Object) that serves as the central wrapper or facade for your application.
How Core-Lib?
Core-Lib
is a plugin and plug-able to otherCore-Libs
.Core-Lib
can discover and merge otherCore-Lib's
configurations.Core-Lib
provides basic/simple/loose tools.Core-Lib
is not delegating third-party libraries.Core-Lib
unit-test the entire library.Core-Lib
deploys anywhere.Core-Lib
recommends architecture and guidelines.
What problems Core-Lib is solving:
Code Chaos:
Ending the cycle of repetitive code creation, ensuring longevity through structured development, and preventing code morphing with each developer’s touch.
Tight Coupling:
Minimizing reliance on a single third-party dependency. Core-lib provides its own unique tools to work with multiple dependencies that are easy to add or remove. Ensuring smoother integration and maintenance of the system.
Testing Trouble:
Unit testing of your entire application, avoiding the need for complicated environments, setups, deployments, dependencies, and steep learning curves. Optimizing the testing process and saving time.
Sluggish Deployment:
Core-Lib is just code that can be run and tested easily anywhere.
Installing
pip install core-lib
Requirements
python > 3.7
Running tests
python -m unittest discover
Example
your_core_lib.yaml
# @package _global_
core_lib:
...
data:
sqlalchemy:
log_queries: false
create_db: true
session:
pool_recycle: 3600
pool_pre_ping: false
url:
protocol: sqlite
...
your_core_lib.yaml
Explained:
your_core_lib.yaml
is the setting for your entire Core-Lib library. The above example will show how to configure core-lib to connect to a database using SQLAlchemy:
-
log_queries
: DefaultFalse
,log_queries
represent and control the echo flag insqlalchemy
. -
create_db
: Create all tables stored in this project-defined entities metadata.1. -
session
: It is a group to configure thesession
settings1. -
url
: Establishing a connection with the database. It can be defined in a single string or a structured format12.
data:
sqlalchemy:
log_queries: false
url:
protocol: postgresql
username: ${oc.env:POSTGRES_USER}
password: ${oc.env:POSTGRES_PASSWORD}
host: ${oc.env:POSTGRES_HOST}
port: ${oc.env:POSTGRES_PORT}
file: ${oc.env:POSTGRES_DB}
your_core_lib.py
from core_lib.core_lib import CoreLib
from core_lib.connection.sql_alchemy_connection_registry import SqlAlchemyConnectionRegistry
class YourCoreLib(CoreLib):
def __init__(self, config: DictConfig):
CoreLib.__init__(self)
db_connection = SqlAlchemyConnectionRegistry(self.config.core_lib.data.sqlalchemy)
self.user = UserDataAccess(db_connection)
...
your_core_lib.py
Explained:
In your_core_lib.py
, a custom CoreLib class
and a SqlAlchemyConnectionRegistry class
are defined to manage database connections.
Defining a new class YourCoreLib that inherits from CoreLib.
- Defining an
__init__
method forYourCoreLib
that takes a config argument of type DictConfig. It is a dictionary type from omegaconf that used by Hydra. - Calling the parent class CoreLib’s
__init__
method usingCoreLib.__init__(self)
to initialize the base class.- Mark core-lib as started
- Enable the use of core-lib observers
- Initialize a
db_connection
object by instantiatingSqlAlchemyConnectionRegistry
with the SQLAlchemy configuration fetched fromself.config.core_lib.data.sqlalchemy
, thereby connecting to and managing the specified database as defined in theyour_core_lib.yaml
configuration file. - Initializing a
UserDataAccess
class responsible for accessing user data from the database, passing thedb_connection
object to it.
From Main
from omegaconf import DictConfig
@hydra.main(config_path='.', config_name='core_lib_config.yaml')
def main(cfg: DictConfig):
your_core_lib = YourCoreLib(cfg)
...
if __name__ == '__main__':
main()
From Main
Explained:
Using the Hydra library to manage configuration for your Core-Lib library.
Decorator: @hydra.main:
-
This decorator tells
Hydra
to use the specifiedYAML
configuration file (core_lib_config.yaml
) located in the root directory of your Core-lib project. -
def main(cfg: DictConfig)
: This is the main function of your Core-Lib library, which takes acfg
config argument of type DictConfig. It is a dictionary type from omegaconf that used by Hydra. -
your_core_lib = YourCoreLib(cfg)
: This line initializes an instance of theYourCoreLib
class (defined earlier) using the configurationcfg
provided byHydra
. This means that your application will use the configuration parameters specified incore_lib_config.yaml
to set up theYourCoreLib
instance. -
main()
: This line calls themain
function when the script is executed directly, starting the execution of your application.
Unit-Test
import unittest
import hydra
from hydra.core.global_hydra import GlobalHydra
def get_config():
GlobalHydra.instance().clear()
hydra.initialize(config_path=os.path.join('..', 'data', 'config'), caller_stack_depth=1)
return hydra.compose('config.yaml')
config = get_config()
class TestCrud(unittest.TestCase):
def setUp(self):
self.your_core_lib = YourCoreLib(config)
def test_your_core_lib(self):
user = self.your_core_lib.user.create({User.name.key: 'John Dow'})
self.assertDictEqual(user, self.your_core_lib.user.get(user[User.id.key]))
Code Explained:
Writing unit tests for your YourCoreLib
class using the unittest
framework, alongside Hydra for configuration management.
- Defining a function
get_config()
to useHydra
for loading testingconfig.yaml
located under the testing folder.- Clearing any existing
Hydra
configuration. - Initializing
Hydra
with a configuration path pointing to../data/config and using config.yaml
. - Composing configuration and returning it.
- Clearing any existing
-
Defining a test case class
TestCrud
that inherits fromunittest.TestCase
.setUp()
: Called before each test method is executed. Initializes an instance ofYourCoreLib
using the configuration obtained earlier and assigns it toself.your_core_lib
.
-
Defining the test method
test_your_core_lib
:- This method tests functionality related to creating and retrieving a user.
- It creates a user using
self.your_core_lib.user.create()
method, passing in a dictionary with user data. - Then, it retrieves the user using
self.your_core_lib.user.get()
method with the user’s ID obtained from the creation step. - Finally, it asserts that the retrieved user matches the created user.
your_core_lib_instance.py
class YourCoreLibInstance(object):
_app_instance = None
@staticmethod
def init(core_lib_cfg):
if not YourCoreLibInstance._app_instance:
YourCoreLibInstance._app_instance = YourCoreLib(core_lib_cfg)
@staticmethod
def get() -> YourCoreLib:
return YourCoreLibInstance._app_instance
Code Explained:
Implementing a singleton pattern ensures consistent usage of the YourCoreLib class across multiple web views and test files, promoting efficiency and consistency.
-
init(core_lib_cfg)
:- This is a static method (decorated with @staticmethod) responsible for initializing the singleton instance of YourCoreLib.
- It takes a
core_lib_cfg
argument, a configuration needed to initializeYourCoreLib
. - If
_app_instance
is not already set (i.e., it’s None), it initializes_app_instance
by creating an instance ofYourCoreLib
with the provided configuration.
-
get() -> YourCoreLib
:- This is another static method responsible for returning the singleton instance of YourCoreLib.
- It simply returns the
_app_instance
, which is thesingleton
instance ofYourCoreLib
.
Flask
view_user.py
from flask import request, Flask
from http import HTTPStatus
from flask import jsonify
from your_core_lib_instance import YourCoreLibInstance
from core_lib.web_helpers.decorators import HandleException
from core_lib.web_helpers.flask.require_login import RequireLogin
from core_lib.web_helpers.request_response_helpers import request_body_dict, response_ok
app = Flask(__name__)
your_core_lib = YourCoreLibInstance.get() # retrieve an instance of YourCoreLib
WebHelpersUtils.init(WebHelpersUtils.ServerType.Flask)
@app.route('/api/update_user', methods=['POST'])
@RequireLogin([])
@HandleException()
def api_update_user():
your_core_lib.user.update(request.user.u_id, request_body_dict(request))
return response_ok()
@app.route('/api/get_user', methods=['GET'])
@RequireLogin([])
@HandleException()
def api_get_user():
user_data = your_core_lib.user.get(request.user.u_id, request_body_dict(request))
return response_json(user_data)
if __name__ == "__main__":
app.run(debug=True)
Code Explained:
Defining API endpoints using Flask
(or a similar framework) to handle requests related to updating a user.
- Importing necessary functions from
core_lib.web_helpers.request_response_helpers
:request_body_dict
: A function that extracts and parses the request body into a dictionary.response_ok
: A function that generates a successful response with an appropriate status code.response_status
: A function that generates a response with a specifiedHTTP
status code.- Getting the
singleton
instance ofYourCoreLib
usingYourCoreLibInstance.get()
. This ensures that you’re using the same instance of YourCoreLib throughout your application.
-
app = Flask(__name__)
: Initialization Flask Application -
WebHelpersUtils.init(WebHelpersUtils.ServerType.Flask)
: Initialization Server type, WebHelpersUtils identifies the type of response if it isFlask / Django
type and returns it. -
Route Definitions
:api_update_user
: HandlesPOST
requests and updates user data usingyour_core_lib.user.update()
method. It returns a successful response usingresponse_ok()
.api_get_user
: HandlesGET
requests and retrieves user data usingyour_core_lib.user.get()
method. It returns the user data as aJSON
response.
-
Both route functions are decorated with
@RequireLogin([])
and@HandleException()
decorators to enforce user authentication and handle exceptions. app.run(debug=True)
: StartFlask
application.
Django
view_user.py
from core_lib.web_helpers.request_response_helpers import request_body_dict, response_ok, response_status
your_core_lib = YourCoreLibInstance.get() # retrieve an instance of YourCoreLib
WebHelpersUtils.init(WebHelpersUtils.ServerType.DJANGO)
@require_POST
@RequireLogin()
@HandleException()
def api_update_user(request):
your_core_lib.user.update(request.user.u_id, request_body_dict(request))
return response_status(HTTPStatus.NO_CONTENT)
@require_GET
@RequireLogin()
@HandleException()
def api_update_user(request):
your_core_lib.user.update(request.user.u_id, request_body_dict(request))
return response_ok()
Code Explained:
- Importing necessary functions from
core_lib.web_helpers.request_response_helpers
:request_body_dict
: A function that extracts and parses the request body into a dictionary.response_ok
: A function that generates a successful response with an appropriate status code.response_status
: A function that generates a response with a specifiedHTTP
status code.- Getting the
singleton
instance ofYourCoreLib
usingYourCoreLibInstance.get()
. This ensures that you’re using the same instance of YourCoreLib throughout your application.
-
WebHelpersUtils.init(WebHelpersUtils.ServerType.Django)
: Initialization Server type, WebHelpersUtils identifies the type of response if it isFlask / Django
type and returns it. - api_update_user (for POST requests):
Decorated with @require_POST
: This decorator ensures that the endpoint only responds toPOST
requests.Decorated with @RequireLogin()
: This decorator ensures that the user must be logged in to access this endpoint.Decorated with @HandleException()
: This decorator handles any exceptions that occur within the endpoint function.- Inside the function,
your_core_lib.user.update()
is called to update the user information using the data from the request body (request_body_dict(request)
). - Finally, it returns a response with
HTTP
status codeNO_CONTENT
using response_status (HTTPStatus.NO_CONTENT
).
- api_update_user (for GET requests):
Decorated with @require_GET
: This decorator ensures that the endpoint only responds toGET
requests.Decorated with @RequireLogin()
: This decorator ensures that the user must be logged in to access this endpoint.Decorated with @HandleException()
: This decorator handles any exceptions that occur within the endpoint function.- Inside the function,
your_core_lib.user.update()
is called to update the user information using the data from the request body (request_body_dict(request)
). - Finally, it returns a successful response using
response_ok()
.
The source
https://github.com/shay-te/core-lib
Example project
https://github.com/shay-te/core-lib/examples
Contributing
Please read CONTRIBUTING.md for details on our code of conduct and the process for submitting pull requests to us.
Authors
Shay Tessler - GitHub
License
This project is licensed under the MIT - see the LICENSE file for details.