Skip to content

Commit

Permalink
docs: 📚 added AuthClient docs, 2-legged oauth example (#7)
Browse files Browse the repository at this point in the history
  • Loading branch information
shimizust authored Feb 14, 2023
1 parent 070df18 commit f8b3180
Show file tree
Hide file tree
Showing 6 changed files with 230 additions and 22 deletions.
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -759,7 +759,7 @@ Base class: [BaseRestliResponse](#class-baserestliresponse)

| Properties | Type | Description |
|---|---|---|
| `status` | int | The status code of the update call |
| `status` | int | The status code of the individual update call |

#### *class BatchDeleteResponse()*

Expand Down
35 changes: 35 additions & 0 deletions examples/oauth_2legged.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
"""
Example calls to fetch a 2-legged access token and introspect the token. A 2-legged access token
obtained using the Client Credentials Flow, allows your application to access APIs that are not member
specific.
Note: By default, developer applications do NOT have the Client Credentials Flow (2-legged) enabled. This
example will only work if your application has had this flow enabled.
"""

import os,sys
sys.path.append(os.path.join(os.path.dirname(__file__), '..'))

from dotenv import load_dotenv,find_dotenv
load_dotenv(find_dotenv())

from linkedin_api.clients.auth.client import AuthClient

CLIENT_ID = os.getenv("CLIENT_ID")
CLIENT_SECRET = os.getenv("CLIENT_SECRET")

auth_client = AuthClient(
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET
)

token_response = auth_client.get_two_legged_access_token()
access_token = token_response.access_token
print(f"Status code: {token_response.status_code}")
print(f"Access token: {access_token}\n")

if access_token:
introspection_response = auth_client.introspect_access_token(access_token)
print("Token introspection details:")
print(f"Auth type: {introspection_response.auth_type}")
print(f"Expires at: {introspection_response.expires_at}")
File renamed without changes.
121 changes: 112 additions & 9 deletions linkedin_api/clients/auth/client.py
Original file line number Diff line number Diff line change
@@ -1,21 +1,60 @@
import requests
from linkedin_api.clients.auth.response_formatter import AccessToken3LResponseFormatter
from linkedin_api.clients.auth.response_formatter import AccessToken2LResponseFormatter, AccessToken3LResponseFormatter, IntrospectTokenResponseFormatter, RefreshTokenExchangeResponseFormatter
import linkedin_api.common.constants as constants
from linkedin_api.common.errors import MissingArgumentError
import linkedin_api.clients.auth.utils.oauth as oauth
from linkedin_api.clients.auth.response import AccessToken3LResponse
from linkedin_api.clients.auth.response import AccessToken2LResponse, AccessToken3LResponse, IntrospectTokenResponse, RefreshTokenExchangeResponse
from typing import Optional, List
from linkedin_api.common.constants import HTTP_METHODS

class AuthClient:
"""
A client for making LinkedIn auth-related calls.
Attributes:
client_id (str): The client ID of the developer application.
client_secret (str): The client secret of the developer application.
redirect_url (Optional[str], optional): The redirect URL. This URL is used in the authorization code flow (3-legged OAuth). Users will be redirected to this URL after authorization. Defaults to None.
session (requests.Session): The session instance used to make requests to the Auth server. Session attributes can be modified, which will affect all requests.
"""
def __init__(self, client_id: str, client_secret: str, redirect_url: Optional[str] = None):
"""
The constructor for the AuthClient class.
Args:
client_id (str): The client ID of the developer application.
client_secret (str): The client secret of the developer application.
redirect_url (Optional[str], optional): The redirect URL. This URL is used in the authorization code flow (3-legged OAuth). Users will be redirected to this URL after authorization. Defaults to None.
"""
self.client_id = client_id
self.client_secret = client_secret
self.redirect_url = redirect_url
self.session = requests.Session()


def generate_member_auth_url(self, scopes: List[str], state: Optional[str] = None) -> str:
"""
Generates the member authorization URL to direct members to. Once redirected, the member will be
presented with LinkedIn's OAuth consent page showing the OAuth scopes your application is requesting
on behalf of the user.
Args:
scopes (List[str]): An array of OAuth scopes (3-legged member permissions) your application is requesting on behalf of the user.
state (Optional[str], optional): An optional string that can be provided to test against CSRF attacks. Defaults to None.
Raises:
MissingArgumentError: Error raised if the auth client was not initialized with a redirect URL, which
is required for this call.
Returns:
str: The member authorization URL
Example:
>>> oauth_url = auth_client.generate_member_auth_url(
scopes=["r_liteprofile", "rw_ads"],
state="abc123"
)
"""
if self.redirect_url is None:
raise MissingArgumentError("The redirect_url is missing from the AuthClient.")

Expand All @@ -26,7 +65,24 @@ def generate_member_auth_url(self, scopes: List[str], state: Optional[str] = Non
state = state
)

def exchange_auth_code_for_access_token(self, code: str):
def exchange_auth_code_for_access_token(self, code: str) -> AccessToken3LResponse:
"""
Exchanges an authorization code for a 3-legged access token. After member authorization,
the browser redirects to the provided redirect URL, setting the authorization code on the
`code` query parameter.
Args:
code (str): The authorization code to exchange for an access token
Returns:
AccessToken3LResponse: An instance of the AccessToken3LResponse class representing the
3-legged access token response details.
Example:
>>> response = auth_client.exchange_auth_code_for_access_token(code=my_auth_code)
>>> access_token = response.access_token
"""

url = f"{constants.OAUTH_BASE_URL}/accessToken"
headers = {
constants.HEADERS.CONTENT_TYPE.value: constants.CONTENT_TYPE.URL_ENCODED.value
Expand All @@ -45,7 +101,23 @@ def exchange_auth_code_for_access_token(self, code: str):

return AccessToken3LResponseFormatter.format_response(response)

def exchange_refresh_token_for_access_token(self, refresh_token: str):
def exchange_refresh_token_for_access_token(self, refresh_token: str) -> RefreshTokenExchangeResponse:
"""
Exchanges a refresh token for a new 3-legged access token. This allows access tokens to be refreshed
without having the member reauthorize your application. Refresh tokens must be enabled for your
application.
Args:
refresh_token (str): The refresh token to exchange for an access token.
Returns:
RefreshTokenExchangeResponse: An instance of RefreshTokenExchangeResponse representing the
refresh token response details.
Example:
>>> response = auth_client.exchange_refresh_token_for_access_token(refresh_token=MY_REFRESH_TOKEN)
>>> access_token = response.access_token
"""
url = f"{constants.OAUTH_BASE_URL}/accessToken"
headers = {
constants.HEADERS.CONTENT_TYPE.value: constants.CONTENT_TYPE.URL_ENCODED.value
Expand All @@ -59,9 +131,23 @@ def exchange_refresh_token_for_access_token(self, refresh_token: str):

request = requests.Request(method=HTTP_METHODS.POST.value, url=url, headers=headers, data=data)
prepared_request = request.prepare()
return self.session.send(prepared_request)
response = self.session.send(prepared_request)
return RefreshTokenExchangeResponseFormatter.format_response(response)

def get_two_legged_access_token(self) -> AccessToken2LResponse:
"""
Use client credential flow (2-legged OAuth) to retrieve a 2-legged access token for accessing
APIs that are not member-specific. Developer applications do not have the client credentials
flow enabled by default.
Returns:
AccessToken2LResponse: An instance of AccessToken2LResponse class representing the two-legged
access token response
def get_two_legged_access_token(self):
Example:
>>> token_response = auth_client.get_two_legged_access_token()
>>> access_token = token_response.access_token
"""
url = f"{constants.OAUTH_BASE_URL}/accessToken"
headers={
constants.HEADERS.CONTENT_TYPE.value: constants.CONTENT_TYPE.URL_ENCODED.value
Expand All @@ -74,9 +160,25 @@ def get_two_legged_access_token(self):

request = requests.Request(method=HTTP_METHODS.POST.value, url=url, headers=headers, data=data)
prepared_request = request.prepare()
return self.session.send(prepared_request)
response = self.session.send(prepared_request)
return AccessToken2LResponseFormatter.format_response(response)

def introspect_access_token(self, access_token: str) -> IntrospectTokenResponse:
"""
Introspect a 2-legged, 3-legged or Enterprise access token to get information on status,
expiry, and other details.
Args:
access_token (str): A 2-legged, 3-legged or Enterprise access token.
def introspect_access_token(self, access_token: str):
Returns:
IntrospectTokenResponse: An instance of IntrospectTokenResponse class representing the
token introspection details
Example:
>>> response = auth_client.introspect_access_token(access_token=MY_ACCESS_TOKEN)
>>> expires_at = response.expires_at
"""
url = f"{constants.OAUTH_BASE_URL}/introspectToken"
headers={
constants.HEADERS.CONTENT_TYPE.value: constants.CONTENT_TYPE.URL_ENCODED.value
Expand All @@ -89,4 +191,5 @@ def introspect_access_token(self, access_token: str):

request = requests.Request(method=HTTP_METHODS.POST.value, url=url, headers=headers, data=data)
prepared_request = request.prepare()
return self.session.send(prepared_request)
response = self.session.send(prepared_request)
return IntrospectTokenResponseFormatter.format_response(response)
92 changes: 81 additions & 11 deletions linkedin_api/clients/auth/response.py
Original file line number Diff line number Diff line change
@@ -1,41 +1,111 @@
from typing import Dict, Any

from linkedin_api.clients.common.response import BaseResponse

class BaseAuthResponse(BaseResponse):
pass

class AccessToken3LResponse(BaseAuthResponse):
def __init__(self, status_code, url, headers, response, access_token, expires_in, refresh_token, refresh_token_expires_in, scope):
super().__init__(status_code=status_code, url=url, headers=headers, response=response)
self.access_token = access_token
self.expires_in = expires_in
self.refresh_token = refresh_token
self.refresh_token_expires_in = refresh_token_expires_in
self.scope = scope
super().__init__(status_code=status_code, url=url, headers=headers, response=response)
self.access_token = access_token
"""
The 3-legged access token.
"""

self.expires_in = expires_in
"""
The TTL for the access token, in seconds.
"""

self.refresh_token = refresh_token
"""
The refresh token value. Only available if refresh tokens are enabled.
"""

self.refresh_token_expires_in = refresh_token_expires_in
"""
The TTL for the refresh token, in seconds. Only available if refresh tokens are enabled.
"""

self.scope = scope
"""
A comma-separated list of scopes authorized by the member (e.g. "r_liteprofile,r_ads").
"""

class AccessToken2LResponse(BaseAuthResponse):
def __init__(self, status_code, url, headers, response, access_token, expires_in):
super().__init__(status_code=status_code, url=url, headers=headers, response=response)
self.access_token = access_token
self.expires_in = expires_in
super().__init__(status_code=status_code, url=url, headers=headers, response=response)
self.access_token = access_token
"""
The two-legged access token.
"""

self.expires_in = expires_in
"""
The TTL of the access token, in seconds.
"""

class IntrospectTokenResponse(BaseAuthResponse):
def __init__(self, status_code, url, headers, response, active, auth_type, authorized_at, client_id, created_at, expires_at, scope, status):
super().__init__(status_code=status_code, url=url, headers=headers, response=response)
self.active = active
"""
Boolean flag whether the token is a valid, active token.
"""

self.auth_type = auth_type
"""
The auth type of the token ("2L", "3L" or "Enterprise_User")
"""

self.authorized_at = authorized_at
"""
Epoch time in seconds, indicating when the token was authorized.
"""

self.client_id = client_id
"""
Developer application client ID.
"""

self.created_at = created_at
"""
Epoch time in seconds, indicating when this token was originally issued.
"""

self.expires_at = expires_at
"""
Epoch time in seconds, indicating when this token will expire.
"""

self.scope = scope
"""
A string containing a comma-separated list of scopes associated with this token. This is only returned for 3-legged member tokens.
"""

self.status = status
"""
The token status ("revoked", "expired", or "active")
"""

class RefreshTokenExchangeResponse(BaseAuthResponse):
def __init__(self, status_code, url, headers, response, access_token, expires_in, refresh_token, refresh_token_expires_in):
super().__init__(status_code=status_code, url=url, headers=headers, response=response)
self.access_token = access_token
"""
The 3-legged access token.
"""

self.expires_in = expires_in
"""
The TTL for the access token, in seconds.
"""

self.refresh_token = refresh_token
"""
The refresh token value.
"""

self.refresh_token_expires_in = refresh_token_expires_in
"""
The TTL for the refresh token, in seconds.
"""
2 changes: 1 addition & 1 deletion linkedin_api/clients/restli/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@

class RestliClient:
"""
A client for making Rest.li API calls.
A client for making Rest.li-based, LinkedIn API calls.
Attributes:
session (requests.Session): The session instance used to send the API requests. Session attributes can
Expand Down

0 comments on commit f8b3180

Please sign in to comment.