
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-Libis a plugin and plug-able to otherCore-Libs.Core-Libcan discover and merge otherCore-Lib'sconfigurations.Core-Libprovides basic/simple/loose tools.Core-Libis not delegating third-party libraries.Core-Libunit-test the entire library.Core-Libdeploys anywhere.Core-Librecommends 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_queriesrepresent 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 thesessionsettings1. -
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_factory import SqlAlchemyConnectionFactory
class YourCoreLib(CoreLib):
def __init__(self, config: DictConfig):
CoreLib.__init__(self)
db_connection = SqlAlchemyConnectionFactory(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 SqlAlchemyConnectionFactory class are defined to manage database connections.
Defining a new class YourCoreLib that inherits from CoreLib.
- Defining an
__init__method forYourCoreLibthat 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_connectionobject by instantiatingSqlAlchemyConnectionFactorywith the SQLAlchemy configuration fetched fromself.config.core_lib.data.sqlalchemy, thereby connecting to and managing the specified database as defined in theyour_core_lib.yamlconfiguration file. - Initializing a
UserDataAccessclass responsible for accessing user data from the database, passing thedb_connectionobject 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
Hydrato use the specifiedYAMLconfiguration 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 acfgconfig 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 theYourCoreLibclass (defined earlier) using the configurationcfgprovided byHydra. This means that your application will use the configuration parameters specified incore_lib_config.yamlto set up theYourCoreLibinstance. -
main(): This line calls themainfunction 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 useHydrafor loading testingconfig.yamllocated under the testing folder.- Clearing any existing
Hydraconfiguration. - Initializing
Hydrawith a configuration path pointing to../data/config and using config.yaml. - Composing configuration and returning it.
- Clearing any existing
-
Defining a test case class
TestCrudthat inherits fromunittest.TestCase.setUp(): Called before each test method is executed. Initializes an instance ofYourCoreLibusing 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_cfgargument, a configuration needed to initializeYourCoreLib. - If
_app_instanceis not already set (i.e., it’s None), it initializes_app_instanceby creating an instance ofYourCoreLibwith 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 thesingletoninstance 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 specifiedHTTPstatus code.- Getting the
singletoninstance ofYourCoreLibusingYourCoreLibInstance.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 / Djangotype and returns it. -
Route Definitions:api_update_user: HandlesPOSTrequests and updates user data usingyour_core_lib.user.update()method. It returns a successful response usingresponse_ok().api_get_user: HandlesGETrequests and retrieves user data usingyour_core_lib.user.get()method. It returns the user data as aJSONresponse.
-
Both route functions are decorated with
@RequireLogin([])and@HandleException()decorators to enforce user authentication and handle exceptions. app.run(debug=True): StartFlaskapplication.
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 specifiedHTTPstatus code.- Getting the
singletoninstance ofYourCoreLibusingYourCoreLibInstance.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 / Djangotype and returns it. - api_update_user (for POST requests):
Decorated with @require_POST: This decorator ensures that the endpoint only responds toPOSTrequests.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
HTTPstatus codeNO_CONTENTusing response_status (HTTPStatus.NO_CONTENT).
- api_update_user (for GET requests):
Decorated with @require_GET: This decorator ensures that the endpoint only responds toGETrequests.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.